JWT Auth

Docs: https://jwt.io

Important

To use jwt you must install 'django-modern-rest[jwt]' extra.

Requiring auth

Note

Current user will always be accessible as self.request.user.

Read more: https://docs.djangoproject.com/en/stable/topics/auth/default/

We provide several classes to require JWT auth in your API, depending on where the token is transferred.

Which one do you need?

	Token in headers	Token in cookies
Classes	HeaderJWTSyncAuth, HeaderJWTAsyncAuth	CookieJWTSyncAuth, CookieJWTAsyncAuth
Best for	Mobile apps, server-to-server calls, and SPAs that keep the token in memory	Browser apps where JavaScript must never touch the token at all
Sent by the browser automatically	No, the client attaches the header itself	Yes, on every matching request
Readable by JavaScript	Yes, the client stores the token itself	No, when the cookie is issued with httponly=True
If your page gets XSS-ed	The token can be read and stolen	The cookie cannot be read, though requests can still be made on the user’s behalf
CSRF	Not applicable	Enforced by us, needs CsrfViewMiddleware
Cross-origin setup	Just send the header	Needs SameSite, Secure, and CORS care

When in doubt, use headers. Reach for cookies when the requirement is specifically “the frontend must not be able to read the token”.

Token in headersToken in cookies

When in doubt, use this as the default way to receive tokens.

Use HeaderJWTSyncAuth for sync views and HeaderJWTAsyncAuth for async views.

They are also available under their older names, JWTSyncAuth and JWTAsyncAuth.

You can customize:

• Security scheme name, default: jwt

• Header name, default: Authorization

• Header value prefix, default: Bearer

• Advanced jwt parameters

Example, how to use the auth class and how to get self.request.user:

views.py

Show imports... 8 lines hidden

[source]

from django.contrib.auth.models import User

from dmr import Controller
from dmr.plugins.pydantic import PydanticSerializer
from dmr.security import AuthenticatedHttpRequest
from dmr.security.jwt import HeaderJWTSyncAuth


class APIController(Controller[PydanticSerializer]):
    request: AuthenticatedHttpRequest[User]
    auth = (HeaderJWTSyncAuth(),)

    def get(self) -> str:
        # Let's test that `User` has the correct type:
        assert self.request.user.is_authenticated
        return 'authed'

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": {
      "jwt": {
        "bearerFormat": "JWT",
        "description": "JWT token auth",
        "scheme": "Bearer",
        "type": "http"
      }
    }
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/apicontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getApicontrollerApiApicontroller",
        "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",
            "headers": {
              "WWW-Authenticate": {
                "description": "Challenges that the client can use to authenticate this request",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "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": [
          {
            "jwt": []
          }
        ]
      }
    }
  }
}

Custom user models are automatically supported.

Tip

Auth classes are tried in order, so you can accept both transports at once with auth = (CookieJWTSyncAuth(), HeaderJWTSyncAuth()). The cookie auth returns None when its cookie is missing, which lets the header auth run next.

Customizing auth

JWT Auth supports a lot of customization options: starting from leeway and claim verification up to the secret key customization.

See decode() for more info on all configuration options.

JSON backend

Token payloads are encoded and decoded with the same JSON backend we use for parsers and renderers: msgspec when it is installed, native pure Python json otherwise. See Alternative JSON backends for the details.

Since JWT auth runs on every authenticated request, having msgspec installed makes encode() and decode() noticeably faster.

Warning

Registered claims are always encoded the same way, but extras can hold arbitrary values. Only json-native values there (str, int, float, bool, None, list, and dict) are guaranteed to produce identical tokens with and without msgspec installed. Other types, like timedelta or set, are either encoded differently or not supported at all by the pure Python fallback. Keep extras json-native when your tokens are issued and verified by different installs.

Reusing pre-existing views

We provide several pre-existing views to get auth tokens. So, users won’t have to write tons of boilerplate code.

JWT with access and refresh tokens

We provide two Reusable controllers to obtain pairs of access and refresh tokens:

1. ObtainTokensSyncController for sync controllers

2. ObtainTokensAsyncController for async controllers

To use them, you will need to:

1. Provide actual types for serializer, request model, and response body

2. Redefine convert_auth_payload() to convert your request model into the kwargs of django.contrib.auth.authenticate() to authenticate your request

3. Redefine make_api_response() to return the response in the format of your choice

views.py

Show imports... 12 lines hidden

[source]

import datetime as dt

from typing_extensions import override

from dmr.plugins.pydantic import PydanticSerializer
from dmr.security.jwt.views import (
    ObtainTokensPayload,
    ObtainTokensResponse,
    ObtainTokensSyncController,
)


# You can also use `ObtainTokensAsyncController` if needed:
class ObtainAccessAndRefreshSyncController(
    ObtainTokensSyncController[
        PydanticSerializer,
        ObtainTokensPayload,
        ObtainTokensResponse,
    ],
):
    @override
    def convert_auth_payload(
        self,
        payload: ObtainTokensPayload,
    ) -> ObtainTokensPayload:
        return payload

    @override
    def make_api_response(self) -> ObtainTokensResponse:
        now = dt.datetime.now(dt.UTC)
        return {
            'access_token': self.create_jwt_token(
                expiration=now + self.jwt_expiration,
                token_type='access',
            ),
            'refresh_token': self.create_jwt_token(
                expiration=now + self.jwt_refresh_expiration,
                token_type='refresh',
            ),
        }

Run result

$ curl http://127.0.0.1:8000/api/auth/ -X POST -d '{"username": "test_user", "password": "password"}' -H 'Content-Type: application/json'
{"access_token":"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxIiwiZXhwIjoxNzg5MTA4MTc2LCJpYXQiOjE3ODkwMjE3NzYsImp0aSI6ImQwY2U4MzI0ZGU4NTRkOTY5NjI0YTE5OTk1MTEyYzQ0IiwiZXh0cmFzIjp7InR5cGUiOiJhY2Nlc3MifX0.FwI49kw8k8MFBlTLJLnVYab0_SEjo5RTcgPGId3lcMo","refresh_token":"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxIiwiZXhwIjoxNzg5ODg1Nzc2LCJpYXQiOjE3ODkwMjE3NzYsImp0aSI6IjExMjY1NTFiYTY4ODRhNzZhOTFlOGUxMWRiNTJkNjM1IiwiZXh0cmFzIjp7InR5cGUiOiJyZWZyZXNoIn19.pfd6XXyTh4GO5oSAB2Kz57zxh0Se67DTXieByOn4TPU"}

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"
      },
      "ObtainTokensPayload": {
        "description": "Payload for default version of a jwt request body.\n\nIs also used as kwargs for :func:`django.contrib.auth.authenticate`.",
        "properties": {
          "password": {
            "title": "Password",
            "type": "string"
          },
          "username": {
            "title": "Username",
            "type": "string"
          }
        },
        "required": [
          "username",
          "password"
        ],
        "title": "ObtainTokensPayload",
        "type": "object"
      },
      "ObtainTokensResponse": {
        "description": "Default response type for refresh token endpoint.",
        "properties": {
          "access_token": {
            "title": "Access Token",
            "type": "string"
          },
          "refresh_token": {
            "title": "Refresh Token",
            "type": "string"
          }
        },
        "required": [
          "access_token",
          "refresh_token"
        ],
        "title": "ObtainTokensResponse",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/obtainaccessandrefreshsynccontroller/": {
      "post": {
        "deprecated": false,
        "operationId": "postObtainaccessandrefreshsynccontrollerApiObtainaccessandrefreshsynccontroller",
        "requestBody": {
          "content": {
            "application/json": {
              "schema": {
                "$ref": "#/components/schemas/ObtainTokensPayload"
              }
            }
          },
          "description": "Payload for default version of a jwt request body.\n\nIs also used as kwargs for :func:`django.contrib.auth.authenticate`.",
          "required": true
        },
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ObtainTokensResponse"
                }
              }
            },
            "description": "OK",
            "headers": {
              "Cache-Control": {
                "description": "Credentials must not be stored in any cache.",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "400": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when request components cannot be parsed"
          },
          "401": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Unauthorized"
          },
          "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"
          }
        },
        "summary": "By default tokens are acquired on post."
      }
    }
  }
}

In this example we utilize pre-defined types of request model and response body, only doing the bare minimum with no customizations.

Things that you can customize:

• Request body format

• Response body format

• JWT settings

• JWT token class to be JWToken subclass with custom logic

• Error messages, see Customizing error messages

• Error handling, see Error handling

• Response status code and any other regular controller or endpoint features

Here’s an example with a lot more customizations:

views.py

Show imports... 11 lines hidden

[source]

import datetime as dt

from typing_extensions import TypedDict, override

from dmr.plugins.pydantic import PydanticSerializer
from dmr.security.jwt.views import (
    ObtainTokensPayload,
    ObtainTokensSyncController,
)


# Custom request and response models:
class _RequestModel(TypedDict):
    email: str
    password: str


class _TokensModel(TypedDict):
    access: str
    refresh: str


class _ResponseModel(TypedDict):
    auth: _TokensModel


# You can also use `ObtainTokensAsyncController` if needed:
class ObtainAccessAndRefreshSyncController(
    ObtainTokensSyncController[
        PydanticSerializer,
        _RequestModel,
        _ResponseModel,
    ],
):
    # Customizes the default jwt settings:
    jwt_issuer = 'my-awesome-company'
    jwt_algorithm = 'HS512'
    jwt_audiences = ('dev', 'qa')

    @override
    def convert_auth_payload(
        self,
        payload: _RequestModel,
    ) -> ObtainTokensPayload:
        return {'username': payload['email'], 'password': payload['password']}

    @override
    def make_api_response(self) -> _ResponseModel:
        now = dt.datetime.now(dt.UTC)
        access = self.create_jwt_token(
            expiration=now + self.jwt_expiration,
            token_type='access',
        )
        refresh = self.create_jwt_token(
            expiration=now + self.jwt_refresh_expiration,
            token_type='refresh',
        )
        return {
            'auth': {
                'access': access,
                'refresh': refresh,
            },
        }

Run result

$ curl http://127.0.0.1:8000/api/auth/ -X POST -d '{"email": "test_user", "password": "password"}' -H 'Content-Type: application/json'
{"auth":{"access":"eyJhbGciOiJIUzUxMiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxIiwiZXhwIjoxNzg5MTA4MTc3LCJpYXQiOjE3ODkwMjE3NzcsImlzcyI6Im15LWF3ZXNvbWUtY29tcGFueSIsImF1ZCI6WyJkZXYiLCJxYSJdLCJqdGkiOiJmZTgxOWYzM2JjYzI0ZGQ2ODkwODkwNmM0OGQzY2VkMiIsImV4dHJhcyI6eyJ0eXBlIjoiYWNjZXNzIn19.pxNXKB3tbIWGtYfooW3DGuQUP1U0U0R9GYsNfuFTcj4nBszxZb_e4UBKO5w-MQG_EFDER6LdEVTS2hCw1XmrYQ","refresh":"eyJhbGciOiJIUzUxMiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxIiwiZXhwIjoxNzg5ODg1Nzc3LCJpYXQiOjE3ODkwMjE3NzcsImlzcyI6Im15LWF3ZXNvbWUtY29tcGFueSIsImF1ZCI6WyJkZXYiLCJxYSJdLCJqdGkiOiI2OGJjZGYyZDNiYmI0MmU1OGQzNjZkMjRjNGE5MGUxOSIsImV4dHJhcyI6eyJ0eXBlIjoicmVmcmVzaCJ9fQ.WBpdMR_aqcNtJwMXzEGKtU_rfLLDxojIVUIEQN-_jyppEMMpBLrMsuZ4oCcS9HHNmB8XVdO4bRpZynjJtDpk-A"}}

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"
      },
      "_RequestModel": {
        "properties": {
          "email": {
            "title": "Email",
            "type": "string"
          },
          "password": {
            "title": "Password",
            "type": "string"
          }
        },
        "required": [
          "email",
          "password"
        ],
        "title": "_RequestModel",
        "type": "object"
      },
      "_ResponseModel": {
        "properties": {
          "auth": {
            "$ref": "#/components/schemas/_TokensModel"
          }
        },
        "required": [
          "auth"
        ],
        "title": "_ResponseModel",
        "type": "object"
      },
      "_TokensModel": {
        "properties": {
          "access": {
            "title": "Access",
            "type": "string"
          },
          "refresh": {
            "title": "Refresh",
            "type": "string"
          }
        },
        "required": [
          "access",
          "refresh"
        ],
        "title": "_TokensModel",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/obtainaccessandrefreshsynccontroller/": {
      "post": {
        "deprecated": false,
        "operationId": "postObtainaccessandrefreshsynccontrollerApiObtainaccessandrefreshsynccontroller",
        "requestBody": {
          "content": {
            "application/json": {
              "schema": {
                "$ref": "#/components/schemas/_RequestModel"
              }
            }
          },
          "required": true
        },
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/_ResponseModel"
                }
              }
            },
            "description": "OK",
            "headers": {
              "Cache-Control": {
                "description": "Credentials must not be stored in any cache.",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "400": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when request components cannot be parsed"
          },
          "401": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Unauthorized"
          },
          "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"
          }
        },
        "summary": "By default tokens are acquired on post."
      }
    }
  }
}

This example also provides issuer and audience in the token, so it can be used together with accepted_issuers and accepted_audiences configurations of HeaderJWTSyncAuth to additionally validate aud and iss JWT token claims.

We want to be sure that this class is at the same time:

1. Easy enough to not write a lot of boilerplate code by default

2. Customizable enough to be able to change a lot of stuff that can be affected by existing business rules

3. Always type safe

Refreshing tokens

Once a user has a refresh token, they can use it to obtain a new pair of access and refresh tokens without re-authenticating. We provide two Reusable controllers for this:

1. RefreshTokenSyncController for sync controllers

2. RefreshTokenAsyncController for async controllers

To use them, you only need to:

1. Provide actual types for serializer, request payload, and response body

2. Redefine convert_refresh_payload() to extract the refresh token string from your request payload

3. Redefine make_api_response() to return the new token pair in the format of your choice

The controller validates that the submitted token:

• Is a valid, non-expired JWT signed with the configured secret

• Has extras.type == 'refresh' (i.e. it is a refresh token, not an access token)

• Belongs to an existing, active user

views.py

Show imports... 12 lines hidden

[source]

import datetime as dt

from typing_extensions import override

from dmr.plugins.pydantic import PydanticSerializer
from dmr.security.jwt.views import (
    ObtainTokensResponse,
    RefreshTokenPayload,
    RefreshTokenSyncController,
)


# You can also use `RefreshTokenAsyncController` if needed:
class RefreshSyncController(
    RefreshTokenSyncController[
        PydanticSerializer,
        RefreshTokenPayload,
        ObtainTokensResponse,
    ],
):
    @override
    def convert_refresh_payload(self, payload: RefreshTokenPayload) -> str:
        return payload['refresh_token']

    @override
    def make_api_response(self) -> ObtainTokensResponse:
        now = dt.datetime.now(dt.UTC)
        return {
            'access_token': self.create_jwt_token(
                expiration=now + self.jwt_expiration,
                token_type='access',
            ),
            'refresh_token': self.create_jwt_token(
                expiration=now + self.jwt_refresh_expiration,
                token_type='refresh',
            ),
        }

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"
      },
      "ObtainTokensResponse": {
        "description": "Default response type for refresh token endpoint.",
        "properties": {
          "access_token": {
            "title": "Access Token",
            "type": "string"
          },
          "refresh_token": {
            "title": "Refresh Token",
            "type": "string"
          }
        },
        "required": [
          "access_token",
          "refresh_token"
        ],
        "title": "ObtainTokensResponse",
        "type": "object"
      },
      "RefreshTokenPayload": {
        "description": "Default request body type for the refresh token endpoint.",
        "properties": {
          "refresh_token": {
            "title": "Refresh Token",
            "type": "string"
          }
        },
        "required": [
          "refresh_token"
        ],
        "title": "RefreshTokenPayload",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/refreshsynccontroller/": {
      "post": {
        "deprecated": false,
        "operationId": "postRefreshsynccontrollerApiRefreshsynccontroller",
        "requestBody": {
          "content": {
            "application/json": {
              "schema": {
                "$ref": "#/components/schemas/RefreshTokenPayload"
              }
            }
          },
          "description": "Default request body type for the refresh token endpoint.",
          "required": true
        },
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ObtainTokensResponse"
                }
              }
            },
            "description": "OK",
            "headers": {
              "Cache-Control": {
                "description": "Credentials must not be stored in any cache.",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "400": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when request components cannot be parsed"
          },
          "401": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Unauthorized"
          },
          "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"
          }
        },
        "summary": "Refresh tokens on POST."
      }
    }
  }
}

Verifying tokens

Sometimes you need a dedicated endpoint to check whether an access token is still valid, without accessing any protected resource. We provide two Reusable controllers for this:

1. VerifyTokenSyncController for sync controllers

2. VerifyTokenAsyncController for async controllers

To use them, you only need to:

1. Provide actual types for serializer and request payload

2. Redefine convert_verify_payload() to extract the access token string from your request payload

The controller validates that the submitted token:

• Is a valid, non-expired JWT signed with the configured secret

• Has extras.type == 'access' (i.e. it is an access token, not a refresh one)

• Belongs to an existing, active user

On success it returns an empty 204 No Content response. On any validation failure it returns 401 Unauthorized.

views.py

Show imports... 6 lines hidden

[source]

from typing_extensions import override

from dmr.plugins.pydantic import PydanticSerializer
from dmr.security.jwt.views import VerifyTokenPayload, VerifyTokenSyncController


# You can also use `VerifyTokenAsyncController` if needed:
class VerifySyncController(
    VerifyTokenSyncController[
        PydanticSerializer,
        VerifyTokenPayload,
    ],
):
    @override
    def convert_verify_payload(self, payload: VerifyTokenPayload) -> str:
        return payload['access_token']

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"
      },
      "VerifyTokenPayload": {
        "description": "Default request body type for the verify token endpoint.",
        "properties": {
          "access_token": {
            "title": "Access Token",
            "type": "string"
          }
        },
        "required": [
          "access_token"
        ],
        "title": "VerifyTokenPayload",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/verifysynccontroller/": {
      "post": {
        "deprecated": false,
        "operationId": "postVerifysynccontrollerApiVerifysynccontroller",
        "requestBody": {
          "content": {
            "application/json": {
              "schema": {
                "$ref": "#/components/schemas/VerifyTokenPayload"
              }
            }
          },
          "description": "Default request body type for the verify token endpoint.",
          "required": true
        },
        "responses": {
          "204": {
            "content": {
              "application/json": {
                "schema": {
                  "type": "null"
                }
              }
            },
            "description": "No Content",
            "headers": {
              "Cache-Control": {
                "description": "Credentials must not be stored in any cache.",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "400": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when request components cannot be parsed"
          },
          "401": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Unauthorized"
          },
          "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"
          }
        },
        "summary": "Verify the token on POST."
      }
    }
  }
}

Issuing tokens as cookies

CookieJWTSyncAuth and CookieJWTAsyncAuth read tokens, but something has to write them first.

Danger

Always set httponly=True and secure=True on these cookies. Without httponly any XSS on your pages can read the token, and without secure it can leak over plain HTTP.

Prefer samesite='strict' (or at least 'lax') and scope the refresh token to the refresh endpoint’s path, so it is never sent to the rest of your API.

To log the user out, set the same cookies to an empty value with max_age=0, which tells the browser to drop them right away. Blocklisting the access token on logout is a good idea too, see the section below.

Blocklisting tokens

Note

Add 'dmr.security.jwt.blocklist' to the INSTALLED_APPS if you want to use tokens blocklist.

JWT tokens might be leaked / outdated / etc. There must be a way to make a valid, non-expired token blocked from auth.

To do so, we provide a default Django app to do so. We store blocked tokens in the database and provide an API to add tokens to the blocklist.

Here’s an example:

views.py

Show imports... 9 lines hidden

[source]

from django.contrib.auth.models import User

from dmr import Controller
from dmr.plugins.pydantic import PydanticSerializer
from dmr.security import AuthenticatedHttpRequest, request_auth
from dmr.security.jwt import HeaderJWTAsyncAuth, request_jwt
from dmr.security.jwt.blocklist import JWTokenBlocklistAsyncMixin


class JWTAuthWithBlocklist(JWTokenBlocklistAsyncMixin, HeaderJWTAsyncAuth):
    """This class will also check that tokens are not blocklisted."""


jwt_blocklist_auth = JWTAuthWithBlocklist()


class APIController(Controller[PydanticSerializer]):
    request: AuthenticatedHttpRequest[User]
    auth = (jwt_blocklist_auth,)

    async def get(self) -> str:
        # Disable tokens for users with old domain emails:
        if self.request.user.email.endswith('@old-domain.com'):
            assert request_auth(self.request) is jwt_blocklist_auth
            await jwt_blocklist_auth.blocklist(
                request_jwt(self.request, strict=True),
            )
        return 'authed'

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": {
      "jwt": {
        "bearerFormat": "JWT",
        "description": "JWT token auth",
        "scheme": "Bearer",
        "type": "http"
      }
    }
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/apicontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getApicontrollerApiApicontroller",
        "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",
            "headers": {
              "WWW-Authenticate": {
                "description": "Challenges that the client can use to authenticate this request",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "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": [
          {
            "jwt": []
          }
        ]
      }
    }
  }
}

We provide two mixin types:

• JWTokenBlocklistAsyncMixin for async auth

• JWTokenBlocklistSyncMixin for sync auth

If this app is installed, we would provide an admin panel by default.

Important

Both mixins add 'jti' to require_claims of the auth class they are mixed into, on top of whatever you pass yourself.

Blocklist rows are keyed by jti, so a token without one can never be blocklisted. Accepting such tokens would mean that the blocklist is silently bypassed: the lookup would match no rows and the token would stay valid forever. We reject them with 401 instead.

If you issue tokens with the controllers we ship, make sure that make_jwt_id() returns a value, our default implementation already does.

Cleaning up expired tokens

The blocklist only answers one question: is this otherwise valid token still allowed? When exp of a token is in the past, decode() rejects it before we even look into the blocklist.

Which means:

• Rows with expires_at in the future must stay, they are the ones actually blocking tokens

• Rows with expires_at in the past can be removed, they cannot change any auth decision anymore

Nothing removes them for us, so the table grows forever while storing rows that can never affect auth again. We recommend deleting them with a periodic job:

myapp/management/commands/cleanup_blocklist.py

Show imports... 8 lines hidden

[source]

import datetime as dt
from typing import Any, final

from django.core.management.base import BaseCommand
from typing_extensions import override

from dmr.security.jwt.blocklist.models import BlocklistedJWToken

#: Must be bigger than the `leeway` of all your jwt auth classes.
GRACE_PERIOD = dt.timedelta(days=1)


@final
class Command(BaseCommand):
    """Delete blocklist entries of tokens that are already expired."""

    help = 'Delete blocklist entries of tokens that are already expired.'

    @override
    def handle(self, *args: Any, **options: Any) -> None:  # noqa: WPS110
        """Expired tokens are rejected by `exp` before the blocklist runs."""
        deleted, _ = BlocklistedJWToken.objects.filter(
            expires_at__lt=dt.datetime.now(dt.UTC) - GRACE_PERIOD,
        ).delete()
        self.stdout.write(f'Removed {deleted} blocklisted tokens')

Then run this task as a periodic job.

Warning

Keep the grace period bigger than the largest leeway you pass to your auth classes. With a non-zero leeway a token is still accepted for that many seconds after exp, and its blocklist row is still doing real work for that long.

Tip

The same reasoning applies to the opaque Token model, see Cleaning up old tokens.

API Reference

class dmr.security.jwt.token.JWToken(sub: str, exp: datetime, iat: datetime = <factory>, iss: str | None = None, aud: str | Sequence[str] | None = None, jti: str | None = None, extras: dict[str, ~typing.Any]=<factory>)

JWT Token DTO.

sub

Subject - usually a unique identifier of the user or equivalent entity.

Type:

str

exp

Expiration - datetime for token expiration.

Type:

datetime.datetime

iat

Issued at - should always be current now.

Type:

datetime.datetime

iss

Issuer - optional unique identifier for the issuer.

Type:

str | None

aud

Audience - intended audience(s).

Type:

str | collections.abc.Sequence[str] | None

jti

JWT ID - a unique identifier of the JWT between different issuers.

Type:

str | None

extras

Extra fields that were found on the JWT token. Only json-native values are guaranteed to be encoded identically with and without msgspec installed.

Type:

dict[str, Any]

Changed in version 0.15.0: Init-only leeway argument was removed. Time-based claims are now validated in a single place for each direction: encode() checks that a token can be issued, while decode() fully relies on pyjwt and its options.

__post_init__() → None

Normalizes datetime claims and runs extra validation.

classmethod decode(encoded_token: str, secret: str, algorithm: str, *, leeway: int = 0, accepted_audiences: str | Sequence[str] | None = None, accepted_issuers: str | Sequence[str] | None = None, require_claims: Sequence[str] | None = None, verify_exp: bool = True, verify_iat: bool = True, verify_jti: bool = True, verify_nbf: bool = True, verify_sub: bool = True, strict_audience: bool = False, enforce_minimum_key_length: bool = True) → Self

Decode a passed in token string and return a Token instance.

Parameters:

• encoded_token – A base64 string containing an encoded JWT.

• secret – The secret with which the JWT is encoded.

• algorithm – The algorithm used to encode the JWT.

• leeway – Number of potential seconds as a clock error for expired tokens.

• accepted_audiences – Verify the audience when decoding the token.

• accepted_issuers – Verify the issuer when decoding the token.

• require_claims – Verify that the given claims are present in the token.

• verify_exp – Verify that the value of the exp (expiration) claim is in the future.

• verify_iat – Verify that iat (issued at) claim value is an integer.

• verify_jti – Check that jti (JWT ID) claim is a string.

• verify_nbf – Verify that the value of the nbf (not before) claim is in the past.

• verify_sub – Check that sub (subject) claim is a string.

• strict_audience – Verify that the value of the aud (audience) claim is a single value, and not a list of values, and matches audience exactly. Requires the value passed to the audience to be a sequence of length 1.

• enforce_minimum_key_length – Raise an auth error when keys are below minimum recommended length.

Returns:

A decoded Token instance.

Raises:

NotAuthenticatedError – If the token is invalid.

See also

https://pyjwt.readthedocs.io/en/stable/api.html#jwt.types.Options

Changed in version 0.15.0: Time-based claims are only validated by pyjwt, we don’t validate them a second time anymore. This means that leeway, verify_exp, and verify_iat are now respected, and that invalid tokens always raise dmr.exceptions.NotAuthenticatedError.

classmethod decode_payload(encoded_token: str, secret: str, algorithms: list[str], *, leeway: int, issuer: str | Sequence[str] | None, audience: str | Sequence[str] | None, options: Options | None) → dict[str, Any]

Decode and verify the JWT and return its payload.

encode(secret: str | bytes, algorithm: str, headers: dict[str, Any] | None = None) → str

Encode the token instance into a string.

Parameters:

• secret – The secret with which the JWT is encoded.

• algorithm – The algorithm used to encode the JWT.

• headers – Optional headers to include in the JWT (e.g., {“kid”: “…”}).

Returns:

An encoded token string.

Raises:

JWTokenError – If the token cannot be issued right now (exp/iat validation) or encoding fails. pyjwt errors are wrapped and the original exception is preserved as the cause.

Changed in version 0.15.0: exp and iat are validated here via validate_issued_claims(), previously it was done during the instance creation. Encoding failures now raise JWTokenError instead of the HTTP-layer InternalServerError.

validate_issued_claims() → None

Ensure that this token makes sense to be issued right now.

Is called by encode(), override it to change or to extend the checks that we run before signing a token.

Raises:

JWTokenError – If this token cannot be issued right now.

Added in version 0.15.0.

final exception dmr.security.jwt.token.JWTokenError

Bases: Exception

Raised when a token cannot be created, encoded, or decoded.

This is a semantic error about the token itself: its claims, the signing algorithm, or the key. It is not an HTTP error, because tokens are regularly created outside of any request: in management commands, background tasks, and scripts.

Added in version 0.15.0.

Header auth

class dmr.security.jwt.auth.HeaderJWTSyncAuth(*, auth_header: str = 'Authorization', auth_scheme: str = 'Bearer', www_authenticate: bool = True, user_id_field: str = 'pk', algorithm: str = 'HS256', security_scheme_name: str = 'jwt', secret: str | None = None, token_cls: type[JWToken] = <class 'dmr.security.jwt.token.JWToken'>, leeway: int = 0, accepted_audiences: str | Sequence[str] | None = None, accepted_issuers: str | Sequence[str] | None = None, require_claims: Sequence[str] | None = None, verify_expiry: bool = True, verify_issued_at: bool = True, verify_jwt_id: bool = True, verify_not_before: bool = True, verify_subject: bool = True, strict_audience: bool = False, enforce_minimum_key_length: bool = True)

Sync jwt auth reading the token from a header.

Defaults to Authorization: Bearer <token>.

Added in version 0.15.0: Previously known as JWTSyncAuth, which is still available as an alias.

__call__(endpoint: Endpoint, controller: Controller[BaseSerializer]) → Self | None

Does check for the correct jwt token.

authenticate(request: HttpRequest, token: JWToken) → AbstractBaseUser

Run all auth pipeline.

check_auth(user: AbstractBaseUser, token: JWToken) → None

Run extra auth checks, raise if something is wrong.

claim_from_token(token: JWToken) → str

Return claim value from the token object.

Override this method if you want to change how claim is extracted from token. For example, if you create email claim, it will be stored in .extras.

So, you would need to use: token.extras['email'].

decode_token(encoded_token: str) → JWToken

Decodes token object from the encoded string.

get_token_from_request(request: HttpRequest) → str | None

Gets the jwt token from the request header.

get_user(token: JWToken) → AbstractBaseUser

Get application user from token.

prepare_token(request: HttpRequest) → JWToken | None

Fetches JWToken instance from the request.

provide_response_specs(metadata: EndpointMetadata, controller_cls: type[Controller[BaseSerializer]], existing_responses: Mapping[HTTPStatus, ResponseSpec]) → list[ResponseSpec]

Provides responses that can happen when user is not authed.

property security_requirement: dict[str, list[str]]

Provides a security schema usage requirement.

property security_schemes: dict[str, SecurityScheme | Reference]

Provides a security schema definition.

set_request_attrs(request: HttpRequest, user: AbstractBaseUser, token: JWToken) → None

Set current user as authed for this request.

split_encoded_token(header: str) → str | None

Splits string like ‘Bearer token’ and returns ‘token’ part.

property www_authenticate_challenge: str | None

Challenge naming the scheme this auth expects, like Bearer.

Returns None for a custom auth_header, because a challenge can only ask the client for the Authorization header. realm is optional for bearer tokens, see RFC 6750 Section 3, and we don’t send it.

class dmr.security.jwt.auth.HeaderJWTAsyncAuth(*, auth_header: str = 'Authorization', auth_scheme: str = 'Bearer', www_authenticate: bool = True, user_id_field: str = 'pk', algorithm: str = 'HS256', security_scheme_name: str = 'jwt', secret: str | None = None, token_cls: type[JWToken] = <class 'dmr.security.jwt.token.JWToken'>, leeway: int = 0, accepted_audiences: str | Sequence[str] | None = None, accepted_issuers: str | Sequence[str] | None = None, require_claims: Sequence[str] | None = None, verify_expiry: bool = True, verify_issued_at: bool = True, verify_jwt_id: bool = True, verify_not_before: bool = True, verify_subject: bool = True, strict_audience: bool = False, enforce_minimum_key_length: bool = True)

Async jwt auth reading the token from a header.

Defaults to Authorization: Bearer <token>.

Added in version 0.15.0: Previously known as JWTAsyncAuth, which is still available as an alias.

async __call__(endpoint: Endpoint, controller: Controller[BaseSerializer]) → Self | None

Does check for the correct jwt token.

async authenticate(request: HttpRequest, token: JWToken) → AbstractBaseUser

Run all auth pipeline.

async check_auth(user: AbstractBaseUser, token: JWToken) → None

Run extra auth checks, raise if something is wrong.

claim_from_token(token: JWToken) → str

Return claim value from the token object.

Override this method if you want to change how claim is extracted from token. For example, if you create email claim, it will be stored in .extras.

So, you would need to use: token.extras['email'].

decode_token(encoded_token: str) → JWToken

Decodes token object from the encoded string.

get_token_from_request(request: HttpRequest) → str | None

Gets the jwt token from the request header.

async get_user(token: JWToken) → AbstractBaseUser

Get application user from token.

prepare_token(request: HttpRequest) → JWToken | None

Fetches JWToken instance from the request.

provide_response_specs(metadata: EndpointMetadata, controller_cls: type[Controller[BaseSerializer]], existing_responses: Mapping[HTTPStatus, ResponseSpec]) → list[ResponseSpec]

Provides responses that can happen when user is not authed.

property security_requirement: dict[str, list[str]]

Provides a security schema usage requirement.

property security_schemes: dict[str, SecurityScheme | Reference]

Provides a security schema definition.

async set_request_attrs(request: HttpRequest, user: AbstractBaseUser, token: JWToken) → None

Set current user as authed for this request.

split_encoded_token(header: str) → str | None

Splits string like ‘Bearer token’ and returns ‘token’ part.

property www_authenticate_challenge: str | None

Challenge naming the scheme this auth expects, like Bearer.

Returns None for a custom auth_header, because a challenge can only ask the client for the Authorization header. realm is optional for bearer tokens, see RFC 6750 Section 3, and we don’t send it.

Note

Since version 0.15.0 JWTSyncAuth and JWTAsyncAuth are kept as aliases of HeaderJWTSyncAuth and HeaderJWTAsyncAuth. Existing code keeps working unchanged. They are soft-deprecated and will be removed before the 1.0.0 release. Do not use them.

Cookie auth

class dmr.security.jwt.auth.CookieJWTSyncAuth(*, cookie_name: str = 'access_token', user_id_field: str = 'pk', algorithm: str = 'HS256', security_scheme_name: str = 'jwt', secret: str | None = None, token_cls: type[JWToken] = <class 'dmr.security.jwt.token.JWToken'>, leeway: int = 0, accepted_audiences: str | Sequence[str] | None = None, accepted_issuers: str | Sequence[str] | None = None, require_claims: Sequence[str] | None = None, verify_expiry: bool = True, verify_issued_at: bool = True, verify_jwt_id: bool = True, verify_not_before: bool = True, verify_subject: bool = True, strict_audience: bool = False, enforce_minimum_key_length: bool = True)

Sync jwt auth reading the token from a cookie.

CSRF is automatically enforced whenever the cookie is present.

Warning

Cookie-based authentication is vulnerable to CSRF attacks in browser-facing contexts. Ensure that django.middleware.csrf.CsrfViewMiddleware is active whenever this auth class is used in a browser-facing application.

Warning

Always issue this cookie with httponly=True and secure=True in production, otherwise the token is readable by any script running on the page and can leak over plain HTTP.

Added in version 0.15.0.

__call__(endpoint: Endpoint, controller: Controller[BaseSerializer]) → Self | None

Enforce CSRF, then authenticate via the cookie token.

authenticate(request: HttpRequest, token: JWToken) → AbstractBaseUser

Run all auth pipeline.

check_auth(user: AbstractBaseUser, token: JWToken) → None

Run extra auth checks, raise if something is wrong.

claim_from_token(token: JWToken) → str

Return claim value from the token object.

Override this method if you want to change how claim is extracted from token. For example, if you create email claim, it will be stored in .extras.

So, you would need to use: token.extras['email'].

decode_token(encoded_token: str) → JWToken

Decodes token object from the encoded string.

get_token_from_request(request: HttpRequest) → str | None

Read the raw jwt token from a cookie.

get_user(token: JWToken) → AbstractBaseUser

Get application user from token.

prepare_token(request: HttpRequest) → JWToken | None

Fetches JWToken instance from the request.

provide_response_specs(metadata: EndpointMetadata, controller_cls: type[Controller[BaseSerializer]], existing_responses: Mapping[HTTPStatus, ResponseSpec]) → list[ResponseSpec]

Declare extra responses for cookie auth + CSRF checks.

property security_requirement: dict[str, list[str]]

Provides a security schema usage requirement.

property security_schemes: dict[str, SecurityScheme | Reference]

Provides a security schema definition.

set_request_attrs(request: HttpRequest, user: AbstractBaseUser, token: JWToken) → None

Set current user as authed for this request.

split_encoded_token(header: str) → str | None

Cookies store the encoded token as-is.

Unlike the Authorization header, a cookie value has no Bearer prefix to strip.

property www_authenticate_challenge: str | None

Cookie auth has no challenge to advertise, so this returns None.

A challenge asks the client for the Authorization header, and this auth reads a cookie instead.

class dmr.security.jwt.auth.CookieJWTAsyncAuth(*, cookie_name: str = 'access_token', user_id_field: str = 'pk', algorithm: str = 'HS256', security_scheme_name: str = 'jwt', secret: str | None = None, token_cls: type[JWToken] = <class 'dmr.security.jwt.token.JWToken'>, leeway: int = 0, accepted_audiences: str | Sequence[str] | None = None, accepted_issuers: str | Sequence[str] | None = None, require_claims: Sequence[str] | None = None, verify_expiry: bool = True, verify_issued_at: bool = True, verify_jwt_id: bool = True, verify_not_before: bool = True, verify_subject: bool = True, strict_audience: bool = False, enforce_minimum_key_length: bool = True)

Async jwt auth reading the token from a cookie.

CSRF is automatically enforced whenever the cookie is present.

Warning

Cookie-based authentication is vulnerable to CSRF attacks in browser-facing contexts. Ensure that django.middleware.csrf.CsrfViewMiddleware is active whenever this auth class is used in a browser-facing application.

Warning

Always issue this cookie with httponly=True and secure=True in production, otherwise the token is readable by any script running on the page and can leak over plain HTTP.

Added in version 0.15.0.

async __call__(endpoint: Endpoint, controller: Controller[BaseSerializer]) → Self | None

Enforce CSRF, then authenticate via the cookie token.

async authenticate(request: HttpRequest, token: JWToken) → AbstractBaseUser

Run all auth pipeline.

async check_auth(user: AbstractBaseUser, token: JWToken) → None

Run extra auth checks, raise if something is wrong.

claim_from_token(token: JWToken) → str

Return claim value from the token object.

Override this method if you want to change how claim is extracted from token. For example, if you create email claim, it will be stored in .extras.

So, you would need to use: token.extras['email'].

decode_token(encoded_token: str) → JWToken

Decodes token object from the encoded string.

get_token_from_request(request: HttpRequest) → str | None

Read the raw jwt token from a cookie.

async get_user(token: JWToken) → AbstractBaseUser

Get application user from token.

prepare_token(request: HttpRequest) → JWToken | None

Fetches JWToken instance from the request.

provide_response_specs(metadata: EndpointMetadata, controller_cls: type[Controller[BaseSerializer]], existing_responses: Mapping[HTTPStatus, ResponseSpec]) → list[ResponseSpec]

Declare extra responses for cookie auth + CSRF checks.

property security_requirement: dict[str, list[str]]

Provides a security schema usage requirement.

property security_schemes: dict[str, SecurityScheme | Reference]

Provides a security schema definition.

async set_request_attrs(request: HttpRequest, user: AbstractBaseUser, token: JWToken) → None

Set current user as authed for this request.

split_encoded_token(header: str) → str | None

Cookies store the encoded token as-is.

Unlike the Authorization header, a cookie value has no Bearer prefix to strip.

property www_authenticate_challenge: str | None

Cookie auth has no challenge to advertise, so this returns None.

A challenge asks the client for the Authorization header, and this auth reads a cookie instead.

Helpers

dmr.security.jwt.auth.request_jwt(request: HttpRequest, *, strict: Literal[True]) → JWToken

dmr.security.jwt.auth.request_jwt(request: HttpRequest, *, strict: bool = False) → JWToken | None

Returns a JWToken from request, if it was authed with it.

When strict is passed and request has no jwt token, we raise AttributeError.

dmr.security.jwt.auth.set_request_attrs(request: HttpRequest, user: AbstractBaseUser, *, token: JWToken | None = None) → None

Set all required properties to the authed request.

Pre-defined views to fetch JWT tokens

class dmr.security.jwt.views.ObtainTokensSyncController(**kwargs)

Sync controller to get access and refresh tokens.

jwt_audiences

String or sequence of string of audiences for JWT token.

jwt_issuer

String of who issued this JWT token.

jwt_algorithm

Default algorithm to use for token signing.

jwt_expiration

Default access token expiration timedelta.

jwt_refresh_expiration

Default refresh token expiration timedelta.

jwt_secret

Alternative token secret for signing. By default uses secret.SECRET_KEY

jwt_token_cls

Possible custom JWT token class.

See also

https://pyjwt.readthedocs.io/en/stable for all the JWT terms and options explanation.

abstractmethod convert_auth_payload(payload: _ObtainTokensT) → ObtainTokensPayload

Convert your custom payload to kwargs that django supports.

See django.contrib.auth.authenticate() docs on which kwargs it supports.

Basically it needs username and password strings.

create_jwt_token(*, expiration: datetime | None = None, token_type: Literal['access', 'refresh'] | None = None, subject: str | None = None, issuer: str | None = None, audiences: str | Sequence[str] | None = None, jwt_id: str | None = None, secret: str | None = None, algorithm: str | None = None, token_headers: dict[str, Any] | None = None) → str

Create correct jwt token of a given expiration and token_type.

login(parsed_body: _ObtainTokensT) → _TokensResponseT

Perform the sync login routine for user.

abstractmethod make_api_response() → _TokensResponseT

Abstract method to create a response payload.

make_jwt_id() → str | None

Create unique token’s jwt id.

post(parsed_body: ~typing.Annotated[~dmr.security.jwt.views._ObtainTokensT, <dmr.components.BodyComponent object at 0x7868ca953c40>]) → _TokensResponseT

By default tokens are acquired on post.

class dmr.security.jwt.views.ObtainTokensAsyncController(**kwargs)

Async controller to get access and refresh tokens.

jwt_audiences

String or sequence of string of audiences for JWT token.

jwt_issuer

String of who issued this JWT token.

jwt_algorithm

Default algorithm to use for token signing.

jwt_expiration

Default token expiration timedelta.

jwt_refresh_expiration

Default refresh token expiration timedelta.

jwt_secret

Alternative token secret for signing. By default uses secret.SECRET_KEY

jwt_token_cls

Possible custom JWT token class.

See also

https://pyjwt.readthedocs.io/en/stable for all the JWT terms and options explanation.

abstractmethod async convert_auth_payload(payload: _ObtainTokensT) → ObtainTokensPayload

Convert your custom payload to kwargs that django supports.

See django.contrib.auth.authenticate() docs on which kwargs it supports.

Basically it needs username and password strings.

create_jwt_token(*, expiration: datetime | None = None, token_type: Literal['access', 'refresh'] | None = None, subject: str | None = None, issuer: str | None = None, audiences: str | Sequence[str] | None = None, jwt_id: str | None = None, secret: str | None = None, algorithm: str | None = None, token_headers: dict[str, Any] | None = None) → str

Create correct jwt token of a given expiration and token_type.

async login(parsed_body: _ObtainTokensT) → _TokensResponseT

Perform the async login routine for user.

abstractmethod async make_api_response() → _TokensResponseT

Abstract method to create a response payload.

make_jwt_id() → str | None

Create unique token’s jwt id.

async post(parsed_body: ~typing.Annotated[~dmr.security.jwt.views._ObtainTokensT, <dmr.components.BodyComponent object at 0x7868ca953c40>]) → _TokensResponseT

By default tokens are acquired on post.

class dmr.security.jwt.views.ObtainTokensPayload

Bases: TypedDict

Payload for default version of a jwt request body.

Is also used as kwargs for django.contrib.auth.authenticate().

class dmr.security.jwt.views.ObtainTokensResponse

Bases: TypedDict

Default response type for refresh token endpoint.

class dmr.security.jwt.views.RefreshTokenSyncController(**kwargs)

Sync controller to refresh access and refresh tokens.

Accepts a refresh token in the request body, validates it, loads the user, and calls make_api_response() to build the response.

jwt_user_id_field

User model field matched against token.sub. Defaults to 'pk'.

jwt_audiences

String or sequence of string of audiences for JWT token.

jwt_issuer

String of who issued this JWT token.

jwt_algorithm

Default algorithm to use for token signing.

jwt_expiration

Default token expiration timedelta.

jwt_refresh_expiration

Default refresh token expiration timedelta.

jwt_secret

Alternative token secret for signing. By default uses secret.SECRET_KEY

jwt_token_cls

Possible custom JWT token class.

check_auth(user: Any) → None

Run extra auth checks, raise if something is wrong.

abstractmethod convert_refresh_payload(payload: _RefreshTokensT) → str

Extract the refresh token string from the request payload.

create_jwt_token(*, expiration: datetime | None = None, token_type: Literal['access', 'refresh'] | None = None, subject: str | None = None, issuer: str | None = None, audiences: str | Sequence[str] | None = None, jwt_id: str | None = None, secret: str | None = None, algorithm: str | None = None, token_headers: dict[str, Any] | None = None) → str

Create correct jwt token of a given expiration and token_type.

abstractmethod make_api_response() → _TokensResponseT

Build the token pair response after a successful refresh.

make_jwt_id() → str | None

Create unique token’s jwt id.

post(parsed_body: ~typing.Annotated[~dmr.security.jwt.views._RefreshTokensT, <dmr.components.BodyComponent object at 0x7868ca953c40>]) → _TokensResponseT

Refresh tokens on POST.

refresh(parsed_body: _RefreshTokensT) → _TokensResponseT

Validate the refresh token, load user, and return new tokens.

class dmr.security.jwt.views.RefreshTokenAsyncController(**kwargs)

Async controller to refresh access and refresh tokens.

Accepts a refresh token in the request body, validates it, loads the user, and calls make_api_response() to build the response.

jwt_user_id_field

User model field matched against token.sub. Defaults to 'pk'.

jwt_audiences

String or sequence of string of audiences for JWT token.

jwt_issuer

String of who issued this JWT token.

jwt_algorithm

Default algorithm to use for token signing.

jwt_expiration

Default token expiration timedelta.

jwt_refresh_expiration

Default refresh token expiration timedelta.

jwt_secret

Alternative token secret for signing. By default uses secret.SECRET_KEY

jwt_token_cls

Possible custom JWT token class.

async check_auth(user: Any) → None

Run extra auth checks, raise if something is wrong.

abstractmethod async convert_refresh_payload(payload: _RefreshTokensT) → str

Extract the refresh token string from the request payload.

create_jwt_token(*, expiration: datetime | None = None, token_type: Literal['access', 'refresh'] | None = None, subject: str | None = None, issuer: str | None = None, audiences: str | Sequence[str] | None = None, jwt_id: str | None = None, secret: str | None = None, algorithm: str | None = None, token_headers: dict[str, Any] | None = None) → str

Create correct jwt token of a given expiration and token_type.

abstractmethod async make_api_response() → _TokensResponseT

Build the token pair response after a successful refresh.

make_jwt_id() → str | None

Create unique token’s jwt id.

async post(parsed_body: ~typing.Annotated[~dmr.security.jwt.views._RefreshTokensT, <dmr.components.BodyComponent object at 0x7868ca953c40>]) → _TokensResponseT

Refresh tokens on POST.

async refresh(parsed_body: _RefreshTokensT) → _TokensResponseT

Validate the refresh token, load user, and return new tokens.

class dmr.security.jwt.views.RefreshTokenPayload

Bases: TypedDict

Default request body type for the refresh token endpoint.

class dmr.security.jwt.views.VerifyTokenSyncController(**kwargs)

Sync controller to verify an access token.

Accepts an access token in the request body, decodes and validates it, ensures it is an access token (not a refresh token), and confirms that the token subject belongs to an existing, active user.

Returns an empty 204 No Content response when the token is valid.

jwt_user_id_field

User model field matched against token.sub. Defaults to 'pk'.

jwt_audiences

String or sequence of string of audiences for JWT token.

jwt_issuer

String of who issued this JWT token.

jwt_algorithm

Default algorithm to use for token signing.

jwt_expiration

Default token expiration timedelta.

jwt_refresh_expiration

Default refresh token expiration timedelta.

jwt_secret

Alternative token secret for signing. By default uses secret.SECRET_KEY

jwt_token_cls

Possible custom JWT token class.

check_auth(user: Any) → None

Run extra checks on the token’s user, raise if something is off.

abstractmethod convert_verify_payload(payload: _VerifyTokenT) → str

Extract the access token string from the request payload.

create_jwt_token(*, expiration: datetime | None = None, token_type: Literal['access', 'refresh'] | None = None, subject: str | None = None, issuer: str | None = None, audiences: str | Sequence[str] | None = None, jwt_id: str | None = None, secret: str | None = None, algorithm: str | None = None, token_headers: dict[str, Any] | None = None) → str

Create correct jwt token of a given expiration and token_type.

get_user(token: JWToken) → AbstractBaseUser

Fetch user by token.

make_jwt_id() → str | None

Create unique token’s jwt id.

post(parsed_body: ~typing.Annotated[~dmr.security.jwt.views._VerifyTokenT, <dmr.components.BodyComponent object at 0x7868ca953c40>]) → None

Verify the token on POST.

verify(parsed_body: _VerifyTokenT) → None

Validate the access token and load its user.

class dmr.security.jwt.views.VerifyTokenAsyncController(**kwargs)

Async controller to verify an access token.

Accepts an access token in the request body, decodes and validates it, ensures it is an access token (not a refresh token), and confirms that the token subject belongs to an existing, active user.

Returns an empty 204 No Content response when the token is valid.

jwt_user_id_field

User model field matched against token.sub. Defaults to 'pk'.

jwt_audiences

String or sequence of string of audiences for JWT token.

jwt_issuer

String of who issued this JWT token.

jwt_algorithm

Default algorithm to use for token signing.

jwt_expiration

Default token expiration timedelta.

jwt_refresh_expiration

Default refresh token expiration timedelta.

jwt_secret

Alternative token secret for signing. By default uses secret.SECRET_KEY

jwt_token_cls

Possible custom JWT token class.

async check_auth(user: Any) → None

Run extra checks on the token’s user, raise if something is off.

abstractmethod async convert_verify_payload(payload: _VerifyTokenT) → str

Extract the access token string from the request payload.

create_jwt_token(*, expiration: datetime | None = None, token_type: Literal['access', 'refresh'] | None = None, subject: str | None = None, issuer: str | None = None, audiences: str | Sequence[str] | None = None, jwt_id: str | None = None, secret: str | None = None, algorithm: str | None = None, token_headers: dict[str, Any] | None = None) → str

Create correct jwt token of a given expiration and token_type.

async get_user(token: JWToken) → AbstractBaseUser

Fetch user by token.

make_jwt_id() → str | None

Create unique token’s jwt id.

async post(parsed_body: ~typing.Annotated[~dmr.security.jwt.views._VerifyTokenT, <dmr.components.BodyComponent object at 0x7868ca953c40>]) → None

Verify the token on POST.

async verify(parsed_body: _VerifyTokenT) → None

Validate the access token and load its user.

class dmr.security.jwt.views.VerifyTokenPayload

Bases: TypedDict

Default request body type for the verify token endpoint.

Blocklist app

final class dmr.security.jwt.blocklist.models.BlocklistedJWToken(*args, **kwargs)

Model for Blocklisted token.

exception DoesNotExist

exception MultipleObjectsReturned

class dmr.security.jwt.blocklist.auth.JWTokenBlocklistSyncMixin(*args, **kwargs)

Sync mixin for working with tokens blocklist.

blocklist(token: JWToken) → tuple[BlocklistedJWToken, bool]

Add token to the blocklist.

check_auth(user: AbstractBaseUser, token: JWToken) → None

Check if the token is in the black list, if so raise the error.

class dmr.security.jwt.blocklist.auth.JWTokenBlocklistAsyncMixin(*args, **kwargs)

Async mixin for working with tokens blocklist.

async blocklist(token: JWToken) → tuple[BlocklistedJWToken, bool]

Add token to the blocklist.

async check_auth(user: AbstractBaseUser, token: JWToken) → None

Check if the token is in the black list, if so raise the error.