Built-in middleware#

CORS#

CORS (Cross-Origin Resource Sharing) is a common security mechanism that is often implemented using middleware. To enable CORS in a litestar application simply pass an instance of CORSConfig to Litestar:

from litestar import Litestar
from litestar.config.cors import CORSConfig

cors_config = CORSConfig(allow_origins=["https://www.example.com"])

app = Litestar(route_handlers=[...], cors_config=cors_config)

CSRF#

CSRF (Cross-site request forgery) is a type of attack where unauthorized commands are submitted from a user that the web application trusts. This attack often uses social engineering that tricks the victim into clicking a URL that contains a maliciously crafted, unauthorized request for a particular Web application. The user’s browser then sends this maliciously crafted request to the targeted Web application. If the user is in an active session with the Web application, the application treats this new request as an authorized request submitted by the user. Thus, the attacker can force the user to perform an action the user didn’t intend, for example:

POST /send-money HTTP/1.1
Host: target.web.app
Content-Type: application/x-www-form-urlencoded

amount=1000usd&to=attacker@evil.com

This middleware prevents CSRF attacks by doing the following:

  1. On the first “safe” request (e.g GET) - set a cookie with a special token created by the server

  2. On each subsequent “unsafe” request (e.g POST) - make sure the request contains either a

    form field or an additional header that has this token (more on this below)

To enable CSRF protection in a Litestar application simply pass an instance of CSRFConfig to the Litestar constructor:

from litestar import Litestar, get, post
from litestar.config.csrf import CSRFConfig


@get()
async def get_resource() -> str:
    # GET is one of the safe methods
    return "some_resource"

@post("{id:int}")
async def create_resource(id: int) -> bool:
    # POST is one of the unsafe methods
    return True

csrf_config = CSRFConfig(secret="my-secret")

app = Litestar([get_resource, create_resource], csrf_config=csrf_config)

The following snippet demonstrates how to change the cookie name to “some-cookie-name” and header name to “some-header-name”.

csrf_config = CSRFConfig(secret="my-secret", cookie_name='some-cookie-name', header_name='some-header-name')

A CSRF protected route can be accessed by any client that can make a request with either the header or form-data key.

Note

The form-data key can not be currently configured. It should only be passed via the key “_csrf_token”

In Python, any client such as requests or httpx can be used. The usage of clients or sessions is recommended due to the cookie persistence it offers across requests. The following is an example using httpx.Client.

import httpx


with httpx.Client() as client:
    get_response = client.get("http://localhost:8000/")

    # "csrftoken" is the default cookie name
    csrf = get_response.cookies["csrftoken"]

    # "x-csrftoken" is the default header name
    post_response_using_header = client.post("http://localhost:8000/", headers={"x-csrftoken": csrf})
    assert post_response_using_header.status_code == 201

    # "_csrf_token" is the default *non* configurable form-data key
    post_response_using_form_data = client.post("http://localhost:8000/1", data={"_csrf_token": csrf})
    assert post_response_using_form_data.status_code == 201

    # despite the header being passed, this request will fail as it does not have a cookie in its session
    # note the usage of ``httpx.post`` instead of ``client.post``
    post_response_with_no_persisted_cookie = httpx.post("http://localhost:8000/1", headers={"x-csrftoken": csrf})
    assert post_response_with_no_persisted_cookie.status_code == 403
    assert "CSRF token verification failed" in post_response_with_no_persisted_cookie.text

Routes can be marked as being exempt from the protection offered by this middleware via handler opts

@post("/post", exclude_from_csrf=True)
def handler() -> None: ...

If you need to exempt many routes at once you might want to consider using the exclude kwarg which accepts list of path patterns to skip in the middleware.

See also

Allowed Hosts#

Another common security mechanism is to require that each incoming request has a “Host” or “X-Forwarded-Host” header, and then to restrict hosts to a specific set of domains - what’s called “allowed hosts”.

Litestar includes an AllowedHostsMiddleware class that can be easily enabled by either passing an instance of AllowedHostsConfig or a list of domains to Litestar:

from litestar import Litestar
from litestar.config.allowed_hosts import AllowedHostsConfig

app = Litestar(
    route_handlers=[...],
    allowed_hosts=AllowedHostsConfig(
        allowed_hosts=["*.example.com", "www.wikipedia.org"]
    ),
)

Note

You can use wildcard prefixes (*.) in the beginning of a domain to match any combination of subdomains. Thus, *.example.com will match www.example.com but also x.y.z.example.com etc. You can also simply put * in trusted hosts, which means allow all. This is akin to turning the middleware off, so in this case it may be better to not enable it in the first place. You should note that a wildcard can only be used only in the prefix of a domain name, not in the middle or end. Doing so will result in a validation exception being raised.

Compression#

HTML responses can optionally be compressed. Litestar has built in support for gzip and brotli. Gzip support is provided through the built-in Starlette classes, and brotli support can be added by installing the brotli extras.

You can enable either backend by passing an instance of CompressionConfig to compression_config of Litestar.

GZIP#

You can enable gzip compression of responses by passing an instance of CompressionConfig with the backend parameter set to "gzip".

You can configure the following additional gzip-specific values:

  • minimum_size: the minimum threshold for response size to enable compression. Smaller responses will not be

    compressed. Defaults is 500, i.e. half a kilobyte.

  • gzip_compress_level: a range between 0-9, see the official python docs.

    Defaults to 9 , which is the maximum value.

from litestar import Litestar
from litestar.config.compression import CompressionConfig

app = Litestar(
    route_handlers=[...],
    compression_config=CompressionConfig(backend="gzip", gzip_compress_level=9),
)

Brotli#

The Brotli package is required to run this middleware. It is available as an extras to litestar with the brotli extra (pip install litestar[brotli]).

You can enable brotli compression of responses by passing an instance of CompressionConfig with the backend parameter set to "brotli".

You can configure the following additional brotli-specific values:

  • minimum_size: the minimum threshold for response size to enable compression. Smaller responses will not be

    compressed. Defaults is 500, i.e. half a kilobyte.

  • brotli_quality: Range [0-11], Controls the compression-speed vs compression-density tradeoff. The higher the

    quality, the slower the compression.

  • brotli_mode: The compression mode can be MODE_GENERIC (default), MODE_TEXT (for UTF-8 format text input), or

    MODE_FONT (for WOFF 2.0).

  • brotli_lgwin: Base 2 logarithm of size. Range is 10 to 24. Defaults to 22.

  • brotli_lgblock: Base 2 logarithm of the maximum input block size. Range is 16 to 24. If set to 0, the value will

    be set based on the quality. Defaults to 0.

  • brotli_gzip_fallback: a boolean to indicate if gzip should be used if brotli is not supported.

from litestar import Litestar
from litestar.config.compression import CompressionConfig

app = Litestar(
    route_handlers=[...],
    compression_config=CompressionConfig(backend="brotli", brotli_gzip_fallback=True),
)

Rate-Limit Middleware#

Litestar includes an optional RateLimitMiddleware that follows the IETF RateLimit draft specification.

To use the rate limit middleware, use the RateLimitConfig:

from litestar import Litestar, MediaType, get
from litestar.middleware.rate_limit import RateLimitConfig

rate_limit_config = RateLimitConfig(rate_limit=("minute", 1), exclude=["/schema"])


@get("/", media_type=MediaType.TEXT, sync_to_thread=False)
def handler() -> str:
    """Handler which should not be accessed more than once per minute."""
    return "ok"


app = Litestar(route_handlers=[handler], middleware=[rate_limit_config.middleware])

The only required configuration kwarg is rate_limit, which expects a tuple containing a time-unit (second, minute, hour, day) and a value for the request quota (integer).

Logging Middleware#

Litestar ships with a robust logging middleware that allows logging HTTP request and responses while building on the logging configuration:

from typing import Dict

from litestar import Litestar, get
from litestar.logging.config import LoggingConfig
from litestar.middleware.logging import LoggingMiddlewareConfig

logging_middleware_config = LoggingMiddlewareConfig()


@get("/", sync_to_thread=False)
def my_handler() -> Dict[str, str]:
    return {"hello": "world"}


app = Litestar(
    route_handlers=[my_handler],
    logging_config=LoggingConfig(),
    middleware=[logging_middleware_config.middleware],
)

from litestar import Litestar, get
from litestar.logging.config import LoggingConfig
from litestar.middleware.logging import LoggingMiddlewareConfig

logging_middleware_config = LoggingMiddlewareConfig()


@get("/", sync_to_thread=False)
def my_handler() -> dict[str, str]:
    return {"hello": "world"}


app = Litestar(
    route_handlers=[my_handler],
    logging_config=LoggingConfig(),
    middleware=[logging_middleware_config.middleware],
)

The logging middleware uses the logger configuration defined on the application level, which allows for using both stdlib logging or structlog , depending on the configuration used (see logging configuration for more details).

Obfuscating Logging Output#

Sometimes certain data, e.g. request or response headers, needs to be obfuscated. This is supported by the middleware configuration:

from litestar.middleware.logging import LoggingMiddlewareConfig

logging_middleware_config = LoggingMiddlewareConfig(
    request_cookies_to_obfuscate={"my-custom-session-key"},
    response_cookies_to_obfuscate={"my-custom-session-key"},
    request_headers_to_obfuscate={"my-custom-header"},
    response_headers_to_obfuscate={"my-custom-header"},
)

The middleware will obfuscate the headers Authorization and X-API-KEY , and the cookie session by default.

Compression and Logging of Response Body#

If both CompressionConfig and LoggingMiddleware have been defined for the application, the response body will be omitted from response logging if it has been compressed, even if "body" has been included in response_log_fields. To force the body of compressed responses to be logged, set include_compressed_body to True , in addition to including "body" in response_log_fields.

Session Middleware#

Litestar includes a SessionMiddleware, offering client- and server-side sessions. Server-side sessions are backed by Litestar’s stores, which offer support for:

  • In memory sessions

  • File based sessions

  • Redis based sessions

Setting up the middleware#

To start using sessions in your application all you have to do is create an instance of a configuration object and add its middleware to your application’s middleware stack:

Hello World#
from os import urandom
from typing import Dict

from litestar import Litestar, Request, delete, get, post
from litestar.middleware.session.client_side import CookieBackendConfig

# we initialize to config with a 16 byte key, i.e. 128 a bit key.
# in real world usage we should inject the secret from the environment
session_config = CookieBackendConfig(secret=urandom(16))  # type: ignore[arg-type]


@get("/session", sync_to_thread=False)
def check_session_handler(request: Request) -> Dict[str, bool]:
    """Handler function that accesses request.session."""
    return {"has_session": request.session != {}}


@post("/session", sync_to_thread=False)
def create_session_handler(request: Request) -> None:
    """Handler to set the session."""
    if not request.session:
        # value can be a dictionary or pydantic model
        request.set_session({"username": "moishezuchmir"})


@delete("/session", sync_to_thread=False)
def delete_session_handler(request: Request) -> None:
    """Handler to clear the session."""
    if request.session:
        request.clear_session()


app = Litestar(
    route_handlers=[check_session_handler, create_session_handler, delete_session_handler],
    middleware=[session_config.middleware],
)
Hello World#
from os import urandom

from litestar import Litestar, Request, delete, get, post
from litestar.middleware.session.client_side import CookieBackendConfig

# we initialize to config with a 16 byte key, i.e. 128 a bit key.
# in real world usage we should inject the secret from the environment
session_config = CookieBackendConfig(secret=urandom(16))  # type: ignore[arg-type]


@get("/session", sync_to_thread=False)
def check_session_handler(request: Request) -> dict[str, bool]:
    """Handler function that accesses request.session."""
    return {"has_session": request.session != {}}


@post("/session", sync_to_thread=False)
def create_session_handler(request: Request) -> None:
    """Handler to set the session."""
    if not request.session:
        # value can be a dictionary or pydantic model
        request.set_session({"username": "moishezuchmir"})


@delete("/session", sync_to_thread=False)
def delete_session_handler(request: Request) -> None:
    """Handler to clear the session."""
    if request.session:
        request.clear_session()


app = Litestar(
    route_handlers=[check_session_handler, create_session_handler, delete_session_handler],
    middleware=[session_config.middleware],
)

Note

Since both client- and server-side sessions rely on cookies (one for storing the actual session data, the other for storing the session ID), they share most of the cookie configuration. A complete reference of the cookie configuration can be found at BaseBackendConfig.

Client-side sessions#

Client side sessions are available through the ClientSideSessionBackend, which offers strong AES-CGM encryption security best practices while support cookie splitting.

Important

ClientSideSessionBackend requires the cryptography library, which can be installed together with litestar as an extra using pip install litestar[cryptography]

cookie_backend.py#
from os import urandom

from litestar import Litestar
from litestar.middleware.session.client_side import CookieBackendConfig

session_config = CookieBackendConfig(secret=urandom(16))  # type: ignore

app = Litestar(middleware=[session_config.middleware])

See also

Server-side sessions#

Server side session store data - as the name suggests - on the server instead of the client. They use a cookie containing a session ID which is a randomly generated string to identify a client and load the appropriate data from the store

from pathlib import Path

from litestar import Litestar
from litestar.middleware.session.server_side import ServerSideSessionConfig
from litestar.stores.file import FileStore

app = Litestar(
    middleware=[ServerSideSessionConfig().middleware],
    stores={"sessions": FileStore(path=Path("session_data"))},
)

See also