Writing your own auth#

We ship auth for the transports most APIs need: HTTP Basic Auth, Django Session Auth, JWT Auth, Token Auth, and django-allauth.

Sooner or later you will need one we don’t ship: credentials that come from a reverse proxy, a signed request, a hardware token, or a legacy scheme some client of yours cannot stop sending. This page shows how to write that auth yourself.

Note

Before writing a new auth class, check whether you only need to change where the credentials come from. Most of our classes let you swap the header, the cookie, or the model without any new code.

The contract#

Your auth class subclasses SyncAuth or AsyncAuth and implements several things:

Member	What it does
`__call__()`	Decides whether this request is authenticated.
`security_schemes()`	Describes the auth itself in the OpenAPI spec.
`security_requirement()`	References that description from every endpoint using this auth.
`www_authenticate_challenge()`	Tells a rejected client how to authenticate.

There is one more optional method: provide_response_specs() declares extra responses your auth can produce. We cover it below.

`X-Forwarded-User` example#

Let’s authenticate requests that come through a reverse proxy which has already checked the user, and passes the username down in a header. This is how oauth2-proxy and most SSO gateways work.

Danger

Only ever do this when the proxy is guaranteed to strip that header from incoming client requests. Otherwise anyone can send X-Forwarded-User: admin and log in as anybody.

auth.py#

[source]

from typing import Final, Self

from django.contrib.auth import get_user_model
from django.contrib.auth.base_user import AbstractBaseUser
from django.core.exceptions import ObjectDoesNotExist
from django.http import HttpRequest
from typing_extensions import override

from dmr import Controller
from dmr.endpoint import Endpoint
from dmr.exceptions import NotAuthenticatedError
from dmr.openapi.objects import Reference, SecurityRequirement, SecurityScheme
from dmr.security import SyncAuth
from dmr.serializer import BaseSerializer

#: Header that our authenticating proxy sets for every request it lets in.
PROXY_USER_HEADER: Final = 'X-Forwarded-User'


class BaseProxyHeaderAuth:
    """Everything that does not depend on sync or async execution."""

    __slots__ = ('header_name', 'security_scheme_name')

    def __init__(
        self,
        *,
        header_name: str = PROXY_USER_HEADER,
        security_scheme_name: str = 'proxy_user',
    ) -> None:
        self.header_name = header_name
        self.security_scheme_name = security_scheme_name

    @property
    def security_schemes(self) -> dict[str, SecurityScheme | Reference]:
        return {
            self.security_scheme_name: SecurityScheme(
                type='apiKey',
                name=self.header_name,
                security_scheme_in='header',
                description='Username of the already authenticated user',
            ),
        }

    @property
    def security_requirement(self) -> SecurityRequirement:
        return {self.security_scheme_name: []}

    @property
    def www_authenticate_challenge(self) -> str | None:
        """
        We read a header of our own, so there is no challenge to send.

        Returning ``None`` here is the whole implementation.
        """

    def get_username(self, request: HttpRequest) -> str | None:
        return request.headers.get(self.header_name)

    def set_request_attrs(
        self,
        request: HttpRequest,
        user: AbstractBaseUser,
    ) -> None:
        request.user = user

        # Needed even for sync auth, so `await request.auser()` keeps
        # working in `sync_to_async` and other mixed contexts:
        async def auser() -> AbstractBaseUser:  # noqa: WPS430
            return user

        request.auser = auser


class ProxyHeaderSyncAuth(BaseProxyHeaderAuth, SyncAuth):
    __slots__ = ()

    @override
    def __call__(
        self,
        endpoint: Endpoint,
        controller: Controller[BaseSerializer],
    ) -> Self | None:
        username = self.get_username(controller.request)
        if not username:
            # The header is missing, so this auth simply does not apply.
            # Return `None`, so the next auth in the chain can try.
            return None
        # The header is here, so the client did mean to use this auth.
        # From now on any problem is an error, not a reason to fall through.
        self.set_request_attrs(controller.request, self.get_user(username))
        return self

    def get_user(self, username: str) -> AbstractBaseUser:
        try:
            return get_user_model().objects.get(
                username=username,
                is_active=True,
            )
        except ObjectDoesNotExist:
            raise NotAuthenticatedError from None

Now use it like any auth we ship:

views.py#

[source]

from django.contrib.auth.models import User

from dmr import Controller
from dmr.plugins.pydantic import PydanticSerializer
from dmr.security import AuthenticatedHttpRequest
from examples.auth.custom.auth import ProxyHeaderSyncAuth


class ProfileController(Controller[PydanticSerializer]):
    request: AuthenticatedHttpRequest[User]
    auth = (ProxyHeaderSyncAuth(),)

    def get(self) -> str:
        username = self.request.user.username
        return f'Hello, {username}'

Run result

$ curl http://127.0.0.1:8000/api/profile/ -X GET -H 'X-Forwarded-User: test_user'
"Hello, test_user"

$ curl http://127.0.0.1:8000/api/profile/ -D - -X GET
HTTP/1.1 401 Unauthorized
date: Mon, 07 Sep 2026 05:54:37 GMT
server: uvicorn
Content-Type: application/json
X-Frame-Options: DENY
Vary: Accept-Language
Content-Language: en
Content-Length: 58
X-Content-Type-Options: nosniff
Referrer-Policy: same-origin
Cross-Origin-Opener-Policy: same-origin

{"detail":[{"msg":"Not authenticated","type":"security"}]}

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "title": "Loc",
            "type": "array"
          },
          "msg": {
            "title": "Msg",
            "type": "string"
          },
          "type": {
            "title": "Type",
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "title": "Detail",
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      }
    },
    "securitySchemes": {
      "proxy_user": {
        "description": "Username of the already authenticated user",
        "in": "header",
        "name": "X-Forwarded-User",
        "type": "apiKey"
      }
    }
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/profilecontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getProfilecontrollerApiProfilecontroller",
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "type": "string"
                }
              }
            },
            "description": "OK"
          },
          "401": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when auth was not successful"
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          }
        },
        "security": [
          {
            "proxy_user": []
          }
        ]
      }
    }
  }
}

Deciding the outcome#

__call__() has four possible outcomes, and picking the right one is the part that is easy to get wrong:

Outcome	What happens
`return self`	Authentication succeeded, we stop and run the endpoint.
`return None`	This auth does not apply, we try the next one in the chain. When it is the last one, the request gets a `401`.
raise `NotAuthenticatedError`	Authentication failed, we stop the chain right there and return a `401`.
raise `APIError`	Same, but with a status code of your choice. Use it for unexpected cases.

The rule of thumb: return None while you still cannot tell whether the client meant to use this auth at all, and raise once you can.

In the example above the header being absent means “not my request”, so we return None. A header with an unknown username means “my request, and it is wrong”, so get_user raises.

Warning

Getting this backwards breaks auth chains in a way that is easy to miss. We shipped that bug ourselves in issue 1289: our cookie-based auth ran its CSRF check before looking at whether its cookie was even present, so a request carrying no cookie at all could never fall through to the next auth in the chain.

If your auth does anything that can fail before it knows the request is meant for it, do that check after you have the credentials.

Describing it in OpenAPI#

The two security_* properties work together:

security_schemes returns the named definitions to publish in components.securitySchemes. A name maps to a SecurityScheme
security_requirement returns the names an endpoint requires, which lands in the operation’s security field

Pick the type that matches your transport. Our example reads its own header, so it is apiKey. Had it read Authorization, it would be type='http' with a scheme, and OpenAPI clients would render a proper login box for it.

Both are properties, not class attributes, because they usually depend on the instance configuration, like the header name above.

Setting the request attributes#

A successful auth is expected to tell the rest of the request who the user is:

request.user is what your endpoints and Django itself read
request.auser is its awaitable counterpart. Set it even in sync auth, so code that crosses the sync / async boundary keeps working

Annotate the controller with AuthenticatedHttpRequest once you set user, and self.request.user becomes properly typed, as in views.py above.

We also store the auth instance itself on the request, so request_auth() can tell you which auth of the chain succeeded. That one we set for you, you don’t have to.

If your auth resolves something else worth keeping, like a token row or a session, store it under a name of your own and give your users a helper to read it back. That is exactly what request_jwt() and request_token() do.

Telling clients how to authenticate#

RFC 9110 Section 15.5.2 wants every 401 to carry a challenge, so www_authenticate_challenge() is abstract: every auth has to answer this question explicitly.

A challenge can only name a scheme the client sends in the Authorization header. Our example reads a header of its own, so it has nothing to advertise and returns None. Read Authorization instead, and you would return the scheme name:

>>> from dmr.security.jwt import HeaderJWTSyncAuth
>>> from examples.auth.custom.auth import ProxyHeaderSyncAuth

>>> HeaderJWTSyncAuth().www_authenticate_challenge
'Bearer'

>>> ProxyHeaderSyncAuth().www_authenticate_challenge is None
True

See the challenges section for what the header looks like and how several auth instances combine.

Sync and async#

Sync and async auth are separate classes: a sync controller cannot use async auth, and the other way around. So an auth meant for both ends up as a pair, with the shared parts in a common base:

async_auth.py#

[source]

from typing import Self

from django.contrib.auth import get_user_model
from django.contrib.auth.base_user import AbstractBaseUser
from django.core.exceptions import ObjectDoesNotExist
from typing_extensions import override

from dmr import Controller
from dmr.endpoint import Endpoint
from dmr.exceptions import NotAuthenticatedError
from dmr.security import AsyncAuth
from dmr.serializer import BaseSerializer
from examples.auth.custom.auth import BaseProxyHeaderAuth


class ProxyHeaderAsyncAuth(BaseProxyHeaderAuth, AsyncAuth):
    __slots__ = ()

    @override
    async def __call__(
        self,
        endpoint: Endpoint,
        controller: Controller[BaseSerializer],
    ) -> Self | None:
        username = self.get_username(controller.request)
        if not username:
            return None
        self.set_request_attrs(
            controller.request,
            await self.get_user(username),
        )
        return self

    async def get_user(self, username: str) -> AbstractBaseUser:
        try:
            return await get_user_model().objects.aget(
                username=username,
                is_active=True,
            )
        except ObjectDoesNotExist:
            raise NotAuthenticatedError from None

Note that only the parts that touch the database differ. Everything else, including the whole OpenAPI description, comes from BaseProxyHeaderAuth.

When you set auth globally and your project has both kinds of endpoints, wrap the pair in SyncOrAsyncAuth:

settings.py#

>>> from dmr.security import SyncOrAsyncAuth
>>> from dmr.settings import Settings
>>> from examples.auth.custom.async_auth import ProxyHeaderAsyncAuth
>>> from examples.auth.custom.auth import ProxyHeaderSyncAuth

>>> DMR_SETTINGS = {
...     Settings.auth: [
...         SyncOrAsyncAuth(
...             ProxyHeaderSyncAuth(),
...             ProxyHeaderAsyncAuth(),
...         ),
...     ],
... }

Extra responses#

Every auth automatically documents the 401 it can produce. If yours can fail in some other way, say so by overriding provide_response_specs(), and the OpenAPI schema will list it:

extra_responses.py#

[source]

from collections.abc import Mapping
from http import HTTPStatus

from typing_extensions import override

from dmr import Controller
from dmr.metadata import EndpointMetadata, ResponseSpec
from dmr.security.base import unauth_response_spec
from dmr.serializer import BaseSerializer
from examples.auth.custom.auth import ProxyHeaderSyncAuth


class SignedProxyHeaderSyncAuth(ProxyHeaderSyncAuth):
    __slots__ = ()

    @override
    def provide_response_specs(
        self,
        metadata: EndpointMetadata,
        controller_cls: type[Controller[BaseSerializer]],
        existing_responses: Mapping[HTTPStatus, ResponseSpec],
    ) -> list[ResponseSpec]:
        return [
            # Keep the `401` that every auth declares:
            *self._add_new_response(
                unauth_response_spec(controller_cls),
                existing_responses,
            ),
            *self._add_new_response(
                ResponseSpec(
                    controller_cls.error_model,
                    status_code=HTTPStatus.FORBIDDEN,
                    description='Raised when the proxy signature is invalid',
                ),
                existing_responses,
            ),
        ]

Two things to notice:

Overriding this replaces the default entirely, so call unauth_response_spec() yourself to keep the 401
_add_new_response is protected on purpose: it is meant for subclasses like yours. It skips a response the endpoint already declares, so you never fight with the endpoint’s own specs

This is how our cookie-based auth documents the 403 that its CSRF check can return.

Rules to follow#

Instances must be stateless#

One instance serves every request. It can live in DMR_SETTINGS and be shared by every controller in the project, across threads and coroutines.

So configuration set in __init__ is fine, but anything per-request is not. Not even a lock. Put per-request data on the request, like set_request_attrs does above.

Define `slots`#

Auth classes are instantiated once but touched on every request, and we enforce __slots__ on our own classes in CI. Declare the attributes on the class that assigns them, and leave __slots__ = () on the subclasses that add none.

Keep credentials out of error reports#

Anything your auth holds in a local variable shows up in tracebacks that error reporting middlewares send to admins. If your auth handles raw secrets, decorate the methods that touch them with django.views.decorators.debug.sensitive_variables().

See the auth views section for the details, they apply to auth classes just as much as to views.

API Reference#

dmr.security.base.unauth_response_spec(controller_cls: type[Controller[BaseSerializer]], metadata: EndpointMetadata | None = None) → ResponseSpec[source]#

Defines the default unauthed response spec.

When metadata is passed and its auth chain can produce a WWW-Authenticate challenge, we also document that header.

Next up#

Once your auth works, everything else we document applies to it unchanged. You will probably want:

How authentication works for how auth is enabled, chained, and disabled
Testing authentication for testing endpoints behind it
OpenAPI for the generated schema

Writing your own auth#

The contract#

X-Forwarded-User example#

Deciding the outcome#

Describing it in OpenAPI#

Setting the request attributes#

Telling clients how to authenticate#

Sync and async#

Extra responses#

Rules to follow#

Instances must be stateless#

Define __slots__#

Keep credentials out of error reports#

API Reference#

Next up#

`X-Forwarded-User` example#

Define `slots`#