JWT Security Backends#
Litestar offers optional JWT based security backends. To use these make sure to install the
pyjwt and cryptography
packages, or simply install Litestar with the jwt
extra:
pip install litestar[jwt]
JWT Auth
Backend#
This is the base JWT Auth backend. You can read about its particular API in the JWTAuth
.
It sends the JWT token using a header - and it expects requests to send the JWT token using the same header key.
Click to see the code
from os import environ
from typing import Any, Dict, Optional
from uuid import UUID
from pydantic import BaseModel, EmailStr
from litestar import Litestar, Request, Response, get, post
from litestar.connection import ASGIConnection
from litestar.openapi.config import OpenAPIConfig
from litestar.security.jwt import JWTAuth, Token
# Let's assume we have a User model that is a pydantic model.
# This though is not required - we need some sort of user class -
# but it can be any arbitrary value, e.g. an SQLAlchemy model, a representation of a MongoDB etc.
class User(BaseModel):
id: UUID
name: str
email: EmailStr
MOCK_DB: Dict[str, User] = {}
# JWTAuth requires a retrieve handler callable that receives the JWT token model and the ASGI connection
# and returns the 'User' instance correlating to it.
#
# Notes:
# - 'User' can be any arbitrary value you decide upon.
# - The callable can be either sync or async - both will work.
async def retrieve_user_handler(token: Token, connection: "ASGIConnection[Any, Any, Any, Any]") -> Optional[User]:
# logic here to retrieve the user instance
return MOCK_DB.get(token.sub)
jwt_auth = JWTAuth[User](
retrieve_user_handler=retrieve_user_handler,
token_secret=environ.get("JWT_SECRET", "abcd123"),
# we are specifying which endpoints should be excluded from authentication. In this case the login endpoint
# and our openAPI docs.
exclude=["/login", "/schema"],
)
# Given an instance of 'JWTAuth' we can create a login handler function:
@post("/login")
async def login_handler(data: User) -> Response[User]:
MOCK_DB[str(data.id)] = data
# you can do whatever you want to update the response instance here
# e.g. response.set_cookie(...)
return jwt_auth.login(identifier=str(data.id), token_extras={"email": data.email}, response_body=data)
# We also have some other routes, for example:
@get("/some-path", sync_to_thread=False)
def some_route_handler(request: "Request[User, Token, Any]") -> Any:
# request.user is set to the instance of user returned by the middleware
assert isinstance(request.user, User)
# request.auth is the instance of 'litestar_jwt.Token' created from the data encoded in the auth header
assert isinstance(request.auth, Token)
# do stuff ...
# We create our OpenAPIConfig as usual - the JWT security scheme will be injected into it.
openapi_config = OpenAPIConfig(
title="My API",
version="1.0.0",
)
# We initialize the app instance and pass the jwt_auth 'on_app_init' handler to the constructor.
# The hook handler will inject the JWT middleware and openapi configuration into the app.
app = Litestar(
route_handlers=[login_handler, some_route_handler],
on_app_init=[jwt_auth.on_app_init],
openapi_config=openapi_config,
)
from os import environ
from typing import Any, Optional
from uuid import UUID
from pydantic import BaseModel, EmailStr
from litestar import Litestar, Request, Response, get, post
from litestar.connection import ASGIConnection
from litestar.openapi.config import OpenAPIConfig
from litestar.security.jwt import JWTAuth, Token
# Let's assume we have a User model that is a pydantic model.
# This though is not required - we need some sort of user class -
# but it can be any arbitrary value, e.g. an SQLAlchemy model, a representation of a MongoDB etc.
class User(BaseModel):
id: UUID
name: str
email: EmailStr
MOCK_DB: dict[str, User] = {}
# JWTAuth requires a retrieve handler callable that receives the JWT token model and the ASGI connection
# and returns the 'User' instance correlating to it.
#
# Notes:
# - 'User' can be any arbitrary value you decide upon.
# - The callable can be either sync or async - both will work.
async def retrieve_user_handler(token: Token, connection: "ASGIConnection[Any, Any, Any, Any]") -> Optional[User]:
# logic here to retrieve the user instance
return MOCK_DB.get(token.sub)
jwt_auth = JWTAuth[User](
retrieve_user_handler=retrieve_user_handler,
token_secret=environ.get("JWT_SECRET", "abcd123"),
# we are specifying which endpoints should be excluded from authentication. In this case the login endpoint
# and our openAPI docs.
exclude=["/login", "/schema"],
)
# Given an instance of 'JWTAuth' we can create a login handler function:
@post("/login")
async def login_handler(data: User) -> Response[User]:
MOCK_DB[str(data.id)] = data
# you can do whatever you want to update the response instance here
# e.g. response.set_cookie(...)
return jwt_auth.login(identifier=str(data.id), token_extras={"email": data.email}, response_body=data)
# We also have some other routes, for example:
@get("/some-path", sync_to_thread=False)
def some_route_handler(request: "Request[User, Token, Any]") -> Any:
# request.user is set to the instance of user returned by the middleware
assert isinstance(request.user, User)
# request.auth is the instance of 'litestar_jwt.Token' created from the data encoded in the auth header
assert isinstance(request.auth, Token)
# do stuff ...
# We create our OpenAPIConfig as usual - the JWT security scheme will be injected into it.
openapi_config = OpenAPIConfig(
title="My API",
version="1.0.0",
)
# We initialize the app instance and pass the jwt_auth 'on_app_init' handler to the constructor.
# The hook handler will inject the JWT middleware and openapi configuration into the app.
app = Litestar(
route_handlers=[login_handler, some_route_handler],
on_app_init=[jwt_auth.on_app_init],
openapi_config=openapi_config,
)
from os import environ
from typing import Any
from uuid import UUID
from pydantic import BaseModel, EmailStr
from litestar import Litestar, Request, Response, get, post
from litestar.connection import ASGIConnection
from litestar.openapi.config import OpenAPIConfig
from litestar.security.jwt import JWTAuth, Token
# Let's assume we have a User model that is a pydantic model.
# This though is not required - we need some sort of user class -
# but it can be any arbitrary value, e.g. an SQLAlchemy model, a representation of a MongoDB etc.
class User(BaseModel):
id: UUID
name: str
email: EmailStr
MOCK_DB: dict[str, User] = {}
# JWTAuth requires a retrieve handler callable that receives the JWT token model and the ASGI connection
# and returns the 'User' instance correlating to it.
#
# Notes:
# - 'User' can be any arbitrary value you decide upon.
# - The callable can be either sync or async - both will work.
async def retrieve_user_handler(token: Token, connection: "ASGIConnection[Any, Any, Any, Any]") -> User | None:
# logic here to retrieve the user instance
return MOCK_DB.get(token.sub)
jwt_auth = JWTAuth[User](
retrieve_user_handler=retrieve_user_handler,
token_secret=environ.get("JWT_SECRET", "abcd123"),
# we are specifying which endpoints should be excluded from authentication. In this case the login endpoint
# and our openAPI docs.
exclude=["/login", "/schema"],
)
# Given an instance of 'JWTAuth' we can create a login handler function:
@post("/login")
async def login_handler(data: User) -> Response[User]:
MOCK_DB[str(data.id)] = data
# you can do whatever you want to update the response instance here
# e.g. response.set_cookie(...)
return jwt_auth.login(identifier=str(data.id), token_extras={"email": data.email}, response_body=data)
# We also have some other routes, for example:
@get("/some-path", sync_to_thread=False)
def some_route_handler(request: "Request[User, Token, Any]") -> Any:
# request.user is set to the instance of user returned by the middleware
assert isinstance(request.user, User)
# request.auth is the instance of 'litestar_jwt.Token' created from the data encoded in the auth header
assert isinstance(request.auth, Token)
# do stuff ...
# We create our OpenAPIConfig as usual - the JWT security scheme will be injected into it.
openapi_config = OpenAPIConfig(
title="My API",
version="1.0.0",
)
# We initialize the app instance and pass the jwt_auth 'on_app_init' handler to the constructor.
# The hook handler will inject the JWT middleware and openapi configuration into the app.
app = Litestar(
route_handlers=[login_handler, some_route_handler],
on_app_init=[jwt_auth.on_app_init],
openapi_config=openapi_config,
)
OAuth2 Bearer
Password Flow#
The OAuth2PasswordBearerAuth
backend inherits from the JWTCookieAuth
backend. It works similarly to the JWTCookieAuth
backend, but is meant to be used for
OAuth 2.0 Bearer password flows.
Click to see the code
from os import environ
from typing import Any, Dict, Optional
from uuid import UUID
from pydantic import BaseModel, EmailStr
from litestar import Litestar, Request, Response, get, post
from litestar.connection import ASGIConnection
from litestar.openapi.config import OpenAPIConfig
from litestar.security.jwt import OAuth2Login, OAuth2PasswordBearerAuth, Token
# Let's assume we have a User model that is a pydantic model.
# This though is not required - we need some sort of user class -
# but it can be any arbitrary value, e.g. an SQLAlchemy model, a representation of a MongoDB etc.
class User(BaseModel):
id: UUID
name: str
email: EmailStr
MOCK_DB: Dict[str, User] = {}
# OAuth2PasswordBearerAuth requires a retrieve handler callable that receives the JWT token model and the ASGI connection
# and returns the 'User' instance correlating to it.
#
# Notes:
# - 'User' can be any arbitrary value you decide upon.
# - The callable can be either sync or async - both will work.
async def retrieve_user_handler(token: "Token", connection: "ASGIConnection[Any, Any, Any, Any]") -> Optional[User]:
# logic here to retrieve the user instance
return MOCK_DB.get(token.sub)
oauth2_auth = OAuth2PasswordBearerAuth[User](
retrieve_user_handler=retrieve_user_handler,
token_secret=environ.get("JWT_SECRET", "abcd123"),
# we are specifying the URL for retrieving a JWT access token
token_url="/login",
# we are specifying which endpoints should be excluded from authentication. In this case the login endpoint
# and our openAPI docs.
exclude=["/login", "/schema"],
)
# Given an instance of 'OAuth2PasswordBearerAuth' we can create a login handler function:
@post("/login")
async def login_handler(request: "Request[Any, Any, Any]", data: "User") -> "Response[OAuth2Login]":
MOCK_DB[str(data.id)] = data
# if we do not define a response body, the login process will return a standard OAuth2 login response. Note the `Response[OAuth2Login]` return type.
# you can do whatever you want to update the response instance here
# e.g. response.set_cookie(...)
return oauth2_auth.login(identifier=str(data.id))
@post("/login_custom")
async def login_custom_response_handler(data: "User") -> "Response[User]":
MOCK_DB[str(data.id)] = data
# you can do whatever you want to update the response instance here
# e.g. response.set_cookie(...)
return oauth2_auth.login(identifier=str(data.id), response_body=data)
# We also have some other routes, for example:
@get("/some-path", sync_to_thread=False)
def some_route_handler(request: "Request[User, Token, Any]") -> Any:
# request.user is set to the instance of user returned by the middleware
assert isinstance(request.user, User)
# request.auth is the instance of 'litestar_jwt.Token' created from the data encoded in the auth header
assert isinstance(request.auth, Token)
# do stuff ...
# We create our OpenAPIConfig as usual - the JWT security scheme will be injected into it.
openapi_config = OpenAPIConfig(
title="My API",
version="1.0.0",
)
# We initialize the app instance and pass the oauth2_auth 'on_app_init' handler to the constructor.
# The hook handler will inject the JWT middleware and openapi configuration into the app.
app = Litestar(
route_handlers=[login_handler, some_route_handler],
on_app_init=[oauth2_auth.on_app_init],
openapi_config=openapi_config,
)
from os import environ
from typing import Any, Optional
from uuid import UUID
from pydantic import BaseModel, EmailStr
from litestar import Litestar, Request, Response, get, post
from litestar.connection import ASGIConnection
from litestar.openapi.config import OpenAPIConfig
from litestar.security.jwt import OAuth2Login, OAuth2PasswordBearerAuth, Token
# Let's assume we have a User model that is a pydantic model.
# This though is not required - we need some sort of user class -
# but it can be any arbitrary value, e.g. an SQLAlchemy model, a representation of a MongoDB etc.
class User(BaseModel):
id: UUID
name: str
email: EmailStr
MOCK_DB: dict[str, User] = {}
# OAuth2PasswordBearerAuth requires a retrieve handler callable that receives the JWT token model and the ASGI connection
# and returns the 'User' instance correlating to it.
#
# Notes:
# - 'User' can be any arbitrary value you decide upon.
# - The callable can be either sync or async - both will work.
async def retrieve_user_handler(token: "Token", connection: "ASGIConnection[Any, Any, Any, Any]") -> Optional[User]:
# logic here to retrieve the user instance
return MOCK_DB.get(token.sub)
oauth2_auth = OAuth2PasswordBearerAuth[User](
retrieve_user_handler=retrieve_user_handler,
token_secret=environ.get("JWT_SECRET", "abcd123"),
# we are specifying the URL for retrieving a JWT access token
token_url="/login",
# we are specifying which endpoints should be excluded from authentication. In this case the login endpoint
# and our openAPI docs.
exclude=["/login", "/schema"],
)
# Given an instance of 'OAuth2PasswordBearerAuth' we can create a login handler function:
@post("/login")
async def login_handler(request: "Request[Any, Any, Any]", data: "User") -> "Response[OAuth2Login]":
MOCK_DB[str(data.id)] = data
# if we do not define a response body, the login process will return a standard OAuth2 login response. Note the `Response[OAuth2Login]` return type.
# you can do whatever you want to update the response instance here
# e.g. response.set_cookie(...)
return oauth2_auth.login(identifier=str(data.id))
@post("/login_custom")
async def login_custom_response_handler(data: "User") -> "Response[User]":
MOCK_DB[str(data.id)] = data
# you can do whatever you want to update the response instance here
# e.g. response.set_cookie(...)
return oauth2_auth.login(identifier=str(data.id), response_body=data)
# We also have some other routes, for example:
@get("/some-path", sync_to_thread=False)
def some_route_handler(request: "Request[User, Token, Any]") -> Any:
# request.user is set to the instance of user returned by the middleware
assert isinstance(request.user, User)
# request.auth is the instance of 'litestar_jwt.Token' created from the data encoded in the auth header
assert isinstance(request.auth, Token)
# do stuff ...
# We create our OpenAPIConfig as usual - the JWT security scheme will be injected into it.
openapi_config = OpenAPIConfig(
title="My API",
version="1.0.0",
)
# We initialize the app instance and pass the oauth2_auth 'on_app_init' handler to the constructor.
# The hook handler will inject the JWT middleware and openapi configuration into the app.
app = Litestar(
route_handlers=[login_handler, some_route_handler],
on_app_init=[oauth2_auth.on_app_init],
openapi_config=openapi_config,
)
from os import environ
from typing import Any
from uuid import UUID
from pydantic import BaseModel, EmailStr
from litestar import Litestar, Request, Response, get, post
from litestar.connection import ASGIConnection
from litestar.openapi.config import OpenAPIConfig
from litestar.security.jwt import OAuth2Login, OAuth2PasswordBearerAuth, Token
# Let's assume we have a User model that is a pydantic model.
# This though is not required - we need some sort of user class -
# but it can be any arbitrary value, e.g. an SQLAlchemy model, a representation of a MongoDB etc.
class User(BaseModel):
id: UUID
name: str
email: EmailStr
MOCK_DB: dict[str, User] = {}
# OAuth2PasswordBearerAuth requires a retrieve handler callable that receives the JWT token model and the ASGI connection
# and returns the 'User' instance correlating to it.
#
# Notes:
# - 'User' can be any arbitrary value you decide upon.
# - The callable can be either sync or async - both will work.
async def retrieve_user_handler(token: "Token", connection: "ASGIConnection[Any, Any, Any, Any]") -> User | None:
# logic here to retrieve the user instance
return MOCK_DB.get(token.sub)
oauth2_auth = OAuth2PasswordBearerAuth[User](
retrieve_user_handler=retrieve_user_handler,
token_secret=environ.get("JWT_SECRET", "abcd123"),
# we are specifying the URL for retrieving a JWT access token
token_url="/login",
# we are specifying which endpoints should be excluded from authentication. In this case the login endpoint
# and our openAPI docs.
exclude=["/login", "/schema"],
)
# Given an instance of 'OAuth2PasswordBearerAuth' we can create a login handler function:
@post("/login")
async def login_handler(request: "Request[Any, Any, Any]", data: "User") -> "Response[OAuth2Login]":
MOCK_DB[str(data.id)] = data
# if we do not define a response body, the login process will return a standard OAuth2 login response. Note the `Response[OAuth2Login]` return type.
# you can do whatever you want to update the response instance here
# e.g. response.set_cookie(...)
return oauth2_auth.login(identifier=str(data.id))
@post("/login_custom")
async def login_custom_response_handler(data: "User") -> "Response[User]":
MOCK_DB[str(data.id)] = data
# you can do whatever you want to update the response instance here
# e.g. response.set_cookie(...)
return oauth2_auth.login(identifier=str(data.id), response_body=data)
# We also have some other routes, for example:
@get("/some-path", sync_to_thread=False)
def some_route_handler(request: "Request[User, Token, Any]") -> Any:
# request.user is set to the instance of user returned by the middleware
assert isinstance(request.user, User)
# request.auth is the instance of 'litestar_jwt.Token' created from the data encoded in the auth header
assert isinstance(request.auth, Token)
# do stuff ...
# We create our OpenAPIConfig as usual - the JWT security scheme will be injected into it.
openapi_config = OpenAPIConfig(
title="My API",
version="1.0.0",
)
# We initialize the app instance and pass the oauth2_auth 'on_app_init' handler to the constructor.
# The hook handler will inject the JWT middleware and openapi configuration into the app.
app = Litestar(
route_handlers=[login_handler, some_route_handler],
on_app_init=[oauth2_auth.on_app_init],
openapi_config=openapi_config,
)
Using a custom token class#
The token class used can be customized with arbitrary fields, by creating a subclass of
Token
, and specifying it on the backend:
import dataclasses
import secrets
from typing import Any, Dict
from litestar import Litestar, Request, get
from litestar.connection import ASGIConnection
from litestar.security.jwt import JWTAuth, Token
@dataclasses.dataclass
class CustomToken(Token):
token_flag: bool = False
@dataclasses.dataclass
class User:
id: str
async def retrieve_user_handler(token: CustomToken, connection: ASGIConnection) -> User:
return User(id=token.sub)
TOKEN_SECRET = secrets.token_hex()
jwt_auth = JWTAuth[User](
token_secret=TOKEN_SECRET,
retrieve_user_handler=retrieve_user_handler,
token_cls=CustomToken,
)
@get("/")
def handler(request: Request[User, CustomToken, Any]) -> Dict[str, Any]:
return {"id": request.user.id, "token_flag": request.auth.token_flag}
app = Litestar(middleware=[jwt_auth.middleware])
import dataclasses
import secrets
from typing import Any
from litestar import Litestar, Request, get
from litestar.connection import ASGIConnection
from litestar.security.jwt import JWTAuth, Token
@dataclasses.dataclass
class CustomToken(Token):
token_flag: bool = False
@dataclasses.dataclass
class User:
id: str
async def retrieve_user_handler(token: CustomToken, connection: ASGIConnection) -> User:
return User(id=token.sub)
TOKEN_SECRET = secrets.token_hex()
jwt_auth = JWTAuth[User](
token_secret=TOKEN_SECRET,
retrieve_user_handler=retrieve_user_handler,
token_cls=CustomToken,
)
@get("/")
def handler(request: Request[User, CustomToken, Any]) -> dict[str, Any]:
return {"id": request.user.id, "token_flag": request.auth.token_flag}
app = Litestar(middleware=[jwt_auth.middleware])
The token will be converted from JSON into the appropriate type, including basic type conversions.
Verifying issuer and audience#
To verify the JWT iss
(issuer) and aud
(audience) claim, a list of accepted
issuers or audiences can bet set on the authentication backend. When a JWT is decoded,
the issuer or audience on the token is compared to the list of accepted issuers /
audiences. If the value in the token does not match any value in the respective list,
a NotAuthorizedException
will be raised, returning a response with a
401 Unauthorized
status.
import dataclasses
import secrets
from typing import Any, Dict
from litestar import Litestar, Request, get
from litestar.connection import ASGIConnection
from litestar.security.jwt import JWTAuth, Token
@dataclasses.dataclass
class User:
id: str
async def retrieve_user_handler(token: Token, connection: ASGIConnection) -> User:
return User(id=token.sub)
jwt_auth = JWTAuth[User](
token_secret=secrets.token_hex(),
retrieve_user_handler=retrieve_user_handler,
accepted_audiences=["https://api.testserver.local"],
accepted_issuers=["https://auth.testserver.local"],
)
@get("/")
def handler(request: Request[User, Token, Any]) -> Dict[str, Any]:
return {"id": request.user.id}
app = Litestar([handler], middleware=[jwt_auth.middleware])
import dataclasses
import secrets
from typing import Any
from litestar import Litestar, Request, get
from litestar.connection import ASGIConnection
from litestar.security.jwt import JWTAuth, Token
@dataclasses.dataclass
class User:
id: str
async def retrieve_user_handler(token: Token, connection: ASGIConnection) -> User:
return User(id=token.sub)
jwt_auth = JWTAuth[User](
token_secret=secrets.token_hex(),
retrieve_user_handler=retrieve_user_handler,
accepted_audiences=["https://api.testserver.local"],
accepted_issuers=["https://auth.testserver.local"],
)
@get("/")
def handler(request: Request[User, Token, Any]) -> dict[str, Any]:
return {"id": request.user.id}
app = Litestar([handler], middleware=[jwt_auth.middleware])
Customizing token validation#
Token decoding / validation can be further customized by overriding the
decode_payload()
method. It will be called by
decode()
with the encoded token string, and must return a
dictionary representing the decoded payload, which will then used by
decode()
to construct an instance of the token class.
import dataclasses
from typing import Any, List, Optional, Sequence, Union
from litestar.security.jwt.token import JWTDecodeOptions, Token
@dataclasses.dataclass
class CustomToken(Token):
@classmethod
def decode_payload(
cls,
encoded_token: str,
secret: str,
algorithms: List[str],
issuer: Optional[List[str]] = None,
audience: Union[str, Sequence[str], None] = None,
options: Optional[JWTDecodeOptions] = None,
) -> Any:
payload = super().decode_payload(
encoded_token=encoded_token,
secret=secret,
algorithms=algorithms,
issuer=issuer,
audience=audience,
options=options,
)
payload["sub"] = payload["sub"].split("@", maxsplit=1)[1]
return payload
import dataclasses
from typing import Any, Optional, Union
from collections.abc import Sequence
from litestar.security.jwt.token import JWTDecodeOptions, Token
@dataclasses.dataclass
class CustomToken(Token):
@classmethod
def decode_payload(
cls,
encoded_token: str,
secret: str,
algorithms: list[str],
issuer: Optional[list[str]] = None,
audience: Union[str, Sequence[str], None] = None,
options: Optional[JWTDecodeOptions] = None,
) -> Any:
payload = super().decode_payload(
encoded_token=encoded_token,
secret=secret,
algorithms=algorithms,
issuer=issuer,
audience=audience,
options=options,
)
payload["sub"] = payload["sub"].split("@", maxsplit=1)[1]
return payload
import dataclasses
from typing import Any
from collections.abc import Sequence
from litestar.security.jwt.token import JWTDecodeOptions, Token
@dataclasses.dataclass
class CustomToken(Token):
@classmethod
def decode_payload(
cls,
encoded_token: str,
secret: str,
algorithms: list[str],
issuer: list[str] | None = None,
audience: str | Sequence[str] | None = None,
options: JWTDecodeOptions | None = None,
) -> Any:
payload = super().decode_payload(
encoded_token=encoded_token,
secret=secret,
algorithms=algorithms,
issuer=issuer,
audience=audience,
options=options,
)
payload["sub"] = payload["sub"].split("@", maxsplit=1)[1]
return payload