Middleware¶

Lilya includes several middleware classes unique to the application but also allowing some other ways of designing them by using protocols.

Lilya middleware¶

The Lilya middleware is the classic already available way of declaring the middleware within an Lilya application.

from lilya.apps import Lilya
from lilya.middleware import DefineMiddleware
from lilya.middleware.trustedhost import TrustedHostMiddleware

app = Lilya(
    routes=[...],
    middleware=[
        DefineMiddleware(
            TrustedHostMiddleware,
            allowed_hosts=["example.com", "*.example.com"],
        ),
        # you can also use import strings
        DefineMiddleware("lilya.middleware.httpsredirect.HTTPSRedirectMiddleware"),
    ],
)

Lilya protocols¶

Lilya protocols are not too different from the Lilya middleware. In fact, the name itself happens only because of the use of the python protocols which forces a certain structure to happen.

When designing a middleware, you can inherit and subclass the MiddlewareProtocol provided by Lilya.

from contextlib import AsyncExitStack
from typing import Optional

from lilya.protocols.middleware import MiddlewareProtocol
from lilya.types import ASGIApp, Receive, Scope, Send


class AsyncExitStackMiddleware(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp"):
        """AsyncExitStack Middleware class.

        Args:
            app: The 'next' ASGI app to call.
            config: The AsyncExitConfig instance.
        """
        super().__init__(app)
        self.app = app

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        if not AsyncExitStack:
            await self.app(scope, receive, send)  # pragma: no cover

        exception: Optional[Exception] = None
        async with AsyncExitStack() as stack:
            scope["lilya_astack"] = stack
            try:
                await self.app(scope, receive, send)
            except Exception as e:
                exception = e
                raise e
        if exception:
            raise exception

MiddlewareProtocol¶

For those coming from a more enforced typed language like Java or C#, a protocol is the python equivalent to an interface.

The MiddlewareProtocol is simply an interface to build middlewares for Lilya by enforcing the implementation of the __init__ and the async def __call__.

Enforcing this protocol also aligns with writing a Pure ASGI Middleware.

Quick sample¶

from typing import Any, Dict

from lilya.protocols.middleware import MiddlewareProtocol
from lilya.types import ASGIApp, Receive, Scope, Send


class SampleMiddleware(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs):
        """SampleMiddleware Middleware class.

        The `app` is always enforced.

        Args:
            app: The 'next' ASGI app to call.
            kwargs: Any arbitrarty data.
        """
        super().__init__(app)
        self.app = app
        self.kwargs = kwargs

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        """
        Implement the middleware logic here
        """
        ...


class AnotherSample(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs: Dict[str, Any]):
        super().__init__(app, **kwargs)
        self.app = app

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        await self.app(scope, receive, send)

Middleware and the application¶

Creating this type of middlewares will make sure the protocols are followed and therefore reducing development errors by removing common mistakes.

To add middlewares to the application is very simple.

Application levelAny other level

from typing import Any, Dict

from lilya.apps import Lilya
from lilya.middleware import DefineMiddleware
from lilya.datastructures import Header
from lilya.protocols.middleware import MiddlewareProtocol
from lilya.types import ASGIApp, Receive, Scope, Send


class SampleMiddleware(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs):
        """SampleMiddleware Middleware class.

        The `app` is always enforced.

        Args:
            app: The 'next' ASGI app to call.
            kwargs: Any arbitrarty data.
        """
        super().__init__(app)
        self.app = app
        self.kwargs = kwargs

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        """
        Implement the middleware logic here
        """
        # optional helper to manipulate/parse the headers and keep them in the scope
        header_instance = Header.ensure_header_instance(scope)
        ...


class AnotherSample(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs: Dict[str, Any]):
        super().__init__(app, **kwargs)
        self.app = app

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        await self.app(scope, receive, send)


app = Lilya(
    routes=[...],
    middleware=[
        DefineMiddleware(SampleMiddleware),
        DefineMiddleware(AnotherSample),
    ],
)

from typing import Any, Dict

from lilya.apps import Lilya
from lilya.middleware import DefineMiddleware
from lilya.protocols.middleware import MiddlewareProtocol
from lilya.routing import Include, Path
from lilya.types import ASGIApp, Receive, Scope, Send


class SampleMiddleware(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs):
        """SampleMiddleware Middleware class.

        The `app` is always enforced.

        Args:
            app: The 'next' ASGI app to call.
            kwargs: Any arbitrarty data.
        """
        super().__init__(app)
        self.app = app
        self.kwargs = kwargs

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        """
        Implement the middleware logic here
        """
        ...


class AnotherSample(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs: Dict[str, Any]):
        super().__init__(app, **kwargs)
        self.app = app

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None: ...


class CustomMiddleware(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp", **kwargs: Dict[str, Any]):
        super().__init__(app, **kwargs)
        self.app = app

    async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None: ...


async def home():
    return "Hello world"


# Via Path
app = Lilya(
    routes=[Path("/", handler=home, middleware=[DefineMiddleware(AnotherSample)])],
    middleware=[DefineMiddleware(SampleMiddleware)],
)


# Via Include
app = Lilya(
    routes=[
        Include(
            "/",
            routes=[Path("/", handler=home, middleware=[DefineMiddleware(SampleMiddleware)])],
            middleware=[DefineMiddleware(CustomMiddleware)],
        )
    ],
    middleware=[DefineMiddleware(AnotherSample)],
)

Quick note¶

Info

The middleware is not limited to Lilya, ChildLilya, Include and Path. We simply choose Path as it looks simpler to read and understand.

Pure ASGI Middleware¶

Lilya follows the ASGI spec. This capability allows for the implementation of ASGI middleware using the ASGI interface directly. This involves creating a chain of ASGI applications that call into the next one. Notably, this approach mirrors the implementation of middleware classes shipped with Lilya.

Example of the most common approach

from lilya.types import ASGIApp, Scope, Receive, Send


class MyMiddleware:
    def __init__(self, app: ASGIApp):
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        await self.app(scope, receive, send)

When implementing a Pure ASGI middleware, it is like implementing an ASGI application, the first parameter should always be an app and the __call__ should always return the app.

BaseAuthMiddleware & AuthenticationMiddleware¶

These are a very special middlewares and and helps with any authentication middleware that can be used within a Lilya application but like everything else, you can design your own.

BaseAuthMiddleware is also an abstract class that simply enforces the implementation of the authenticate method and assigning the result object into a tuple[AuthCredentials | None, UserInterface | None] or None and make it available on every request.

AuthenticationMiddleware is an implementation using backends and most people will prefer it. See Authentication for more details.

Example of a JWT middleware class¶

/src/middleware/jwt.py

from myapp.models import User
from myapp.security.jwt.token import Token
from saffier.exceptions import ObjectNotFound

from lilya._internal._connection import Connection
from lilya.exceptions import NotAuthorized
from lilya.authentication import AuthResult, AuthCredentials
from lilya.middleware.authentication import BaseAuthMiddleware
from lilya.types import ASGIApp


class JWTAuthMiddleware(BaseAuthMiddleware):
    """
    An example how to integrate and design a JWT authentication
    middleware assuming a `myapp` in Lilya.
    """

    def __init__(
        self,
        app: ASGIApp,
        signing_key: str,
        algorithm: str,
        api_key_header: str,
    ):
        super().__init__(app)
        self.app = app
        self.signing_key = signing_key
        self.algorithm = algorithm
        self.api_key_header = api_key_header

    async def retrieve_user(self, user_id) -> User:
        try:
            return await User.get(pk=user_id)
        except ObjectNotFound:
            raise NotAuthorized()

    async def authenticate(self, request: Connection) -> AuthResult:
        token = request.headers.get(self.api_key_header)

        if not token:
            raise NotAuthorized("JWT token not found.")

        token = Token.decode(token=token, key=self.signing_key, algorithm=self.algorithm)

        user = await self.retrieve_user(token.sub)
        return AuthCredentials(), user

Import the BaseAuthMiddleware from lilya.middleware.authentication.
Implement the authenticate and return tuple[AuthCredentials, UserInterface] (AuthResult) or None or raise.

Import the middleware into a Lilya application¶

From the application instanceFrom the settings

from lilya.apps import Lilya
from lilya.middleware import DefineMiddleware
from .middleware.jwt import JWTAuthMiddleware


app = Lilya(routes=[...], middleware=[DefineMiddleware(JWTAuthMiddleware)])

from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware


class AppSettings(Settings):

    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            # you can also use absolute import strings
            DefineMiddleware("project.middleware.jwt.JWTAuthMiddleware")
        ]

# load the settings via LILYA_SETTINGS_MODULE=src.configs.live.AppSettings
app = Lilya(routes=[...])

Tip

To know more about loading the settings and the available properties, have a look at the settings docs.

Middleware and the settings¶

One of the advantages of Lilya is leveraging the settings to make the codebase tidy, clean and easy to maintain. As mentioned in the settings document, the middleware is one of the properties available to use to start a Lilya application.

src/configs/live.py

from __future__ import annotations

from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.compression import GZipMiddleware
from lilya.middleware.httpsredirect import HTTPSRedirectMiddleware


class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        """
        All the middlewares to be added when the application starts.
        """
        return [
            DefineMiddleware(HTTPSRedirectMiddleware),
            DefineMiddleware(GZipMiddleware, minimum_size=500, compresslevel=9),
        ]

Start the application with the new settings

LILYA_SETTINGS_MODULE=configs.live.AppSettings uvicorn src:app

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [28720]
INFO:     Started server process [28722]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

Attention

If LILYA_SETTINGS_MODULE is not specified as the module to be loaded, Lilya will load the default settings but your middleware will not be initialized.

Important¶

If you need to specify parameters in your middleware then you will need to wrap it in a lilya.middleware.DefineMiddleware object to do it so. See GZipMiddleware example.

Available middlewares¶

CSRFMiddleware - Handles with the CSRF.
CORSMiddleware - Handles with the CORS.
TrustedHostMiddleware - Restricts the hosts used for connecting if a given allowed_hosts is populated. Optionally just provide a host_is_trusted parameter in the scope.
TrustedReferrerMiddleware - Handles with the CORS if a given allowed_hosts is populated.
GZipMiddleware - Compression middleware gzip.
HTTPSRedirectMiddleware - Middleware that handles HTTPS redirects for your application. Very useful to be used for production or production like environments.
SessionMiddleware - Middleware that handles the sessions.
SessionFixingMiddleware - Middleware that fixes sessions to client ips.
WSGIMiddleware - Allows to connect WSGI applications and run them inside Lilya. A great example how to use it is available.
XFrameOptionsMiddleware - Middleware that handles specifically against clickjacking.
SecurityMiddleware - Provides several security enhancements to the request/response cycle and adds security headers to the response.
ClientIPMiddleware - Provides facilities to retrieve the client ip. This can be useful for ratelimits.
GlobalContextMiddleware - Allows the use of the [g](./context.md#the-g-object) across request contexts. For Lilya by default active.
LifespanGlobalContextMiddleware - Allows the use of the [g](./context.md#the-g-object) in lifespan requests. For Lilya by default active.
RequestContextMiddleware - Adds a request_context object context without the need to use handlers.
AuthenticationMiddleware & BaseAuthMiddleware - See above.
SessionContextMiddleware- Adds a session object context to be accessed in the handlers or request context in general.

CSRFMiddleware¶

The default parameters used by the CSRFMiddleware are secure by default. Lilya lets you tune how CSRF protection works depending on your needs XHR/fetch or traditional HTML forms without extra dependencies.

from __future__ import annotations

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.csrf import CSRFMiddleware

routes = [...]

# Option one
middleware = [DefineMiddleware(CSRFMiddleware, secret="your-long-unique-secret")]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(CSRFMiddleware, secret="your-long-unique-secret"),
        ]

Check the security section

If you want more details specicially about this, check the Security with CSRF documentation section.

What it protects and how¶

CSRFMiddleware implements the double‑submit cookie pattern:

On safe methods (default: GET, HEAD) it sets a CSRF cookie (default name: csrftoken) if missing.
On unsafe methods (POST, PUT, PATCH, DELETE) it validates an incoming token from:
1. The request header X-CSRFToken (preferred for XHR/fetch), or
2. A form field (default name csrf_token) in application/x-www-form-urlencoded or multipart/form-data bodies.

The value from (1) or (2) must match the signed cookie value.

Note

The middleware reads form bodies only when needed (unsafe methods, header missing, content type is form). It buffers and replays the body to the next app, so your handlers can still call await request.form() or await request.body() normally.

Behaviour at a glance¶

Safe methods: Middleware injects the cookie (if missing) and forwards the response.
Unsafe methods:
1. If the header token is present: Validate vs Cookie.
2. Else if content type is application/x-www-form-urlencoded or multipart/form-data: read, extract form_field_name, replay body to the app, then validate.
3. Otherwise: 403 PermissionDenied.

Examples¶

1. Classic HTML form (no JavaScript)¶

Render the CSRF cookie value into a hidden field. Make sure httponly=False so the template can access the cookie.

Middleware (typical):

DefineMiddleware(
    CSRFMiddleware,
    secret="your-long-unique-secret",
    httponly=False,     # templates read the cookie
    secure=True,        # production
    samesite="lax",
)

2. XHR / fetch (header path)¶

Use the cookie value to set the header. (In SPA/PJAX contexts this is often more convenient.)

async function postData(url, data) {
  const csrf = document.cookie
    .split("; ")
    .find((c) => c.startsWith("csrftoken="))
    ?.split("=")[1];

  const res = await fetch(url, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "X-CSRFToken": csrf,           // header token
    },
    body: JSON.stringify(data),
    credentials: "same-origin",
  });

  return res.json();
}

No need to change the middleware defaults, this uses the header path.

3. File uploads (multipart)¶

Works out of the box with hidden fields in multipart/form-data forms.

4. Custom form field name¶

Prefer a different field name (e.g., csrfmiddlewaretoken)?

DefineMiddleware(
    CSRFMiddleware,
    secret="your-long-unique-secret",
    form_field_name="csrfmiddlewaretoken",
    httponly=False,
)

5. Tightening body buffering¶

If you only accept small forms, reduce the max_body_size:

DefineMiddleware(
    CSRFMiddleware,
    secret="your-long-unique-secret",
    max_body_size=64 * 1024,  # 64 KiB
)

When the body exceeds this limit, the middleware won't parse fallback tokens and will fail CSRF unless the header is present.

Minimal, complete examples¶

Using headers

from lilya.apps import Lilya
from lilya.middleware import DefineMiddleware
from lilya.middleware.csrf import CSRFMiddleware
from lilya.responses import Ok
from lilya.routing import Path

async def login():
    return Ok({"status": "ok"})

app = Lilya(
    routes=[Path("/login", login, methods=["GET", "POST"])],
    middleware=[
        DefineMiddleware(
            CSRFMiddleware,
            secret="change-me-long-random",
            # Using header path; HttpOnly can stay True if you never read the cookie in templates
            httponly=True,
            samesite="lax",
            secure=False,  # True in production
        )
    ],
)

Using a form

from lilya.apps import Lilya
from lilya.middleware import DefineMiddleware
from lilya.middleware.csrf import CSRFMiddleware
from lilya.responses import HTMLResponse, Ok
from lilya.routing import Path
from lilya.requests import Request

# Simple template function; in real apps you might use Jinja.
def login_page(request: Request) -> str:
    token = request.cookies.get("csrftoken", "")  # needs httponly=False
    return f"""
    /login
      <input type="text" name="username" />
      <input type="password" name="password" />
      <input type="hidden" name="csrf_token" value="{token}" />
      <button type="submit">Login</button>
    </form>
    """

async def get_login(request: Request):
    return HTMLResponse(login_page(request))

async def post_login(request: Request):
    form = await request.form()
    return Ok({"username": form.get("username")})

app = Lilya(
    routes=[
        Path("/login", get_login, methods=["GET"]),
        Path("/login", post_login, methods=["POST"]),
    ],
    middleware=[
        DefineMiddleware(
            CSRFMiddleware,
            secret="change-me-long-random",
            httponly=False,   # template reads the cookie
            samesite="lax",
            secure=False,     # True in production
        )
    ],
)

Using a custom field

from lilya.apps import Lilya
from lilya.middleware import DefineMiddleware
from lilya.middleware.csrf import CSRFMiddleware
from lilya.responses import HTMLResponse, Ok
from lilya.routing import Path
from lilya.requests import Request

# Simple template function; in real apps you might use Jinja.
def login_page(request: Request) -> str:
    token = request.cookies.get("csrftoken", "")  # needs httponly=False
    return f"""
    /login
      <input type="text" name="username" />
      <input type="password" name="password" />
      <input type="hidden" name="csrf_token" value="{token}" />
      <button type="submit">Login</button>
    </form>
    """

async def get_login(request: Request):
    return HTMLResponse(login_page(request))

async def post_login(request: Request):
    form = await request.form()
    return Ok({"username": form.get("username")})

app = Lilya(
    routes=[
        Path("/login", get_login, methods=["GET"]),
        Path("/login", post_login, methods=["POST"]),
    ],
    middleware=[
        DefineMiddleware(
            CSRFMiddleware,
            secret="change-me-long-random",
            httponly=False,   # template reads the cookie
            samesite="lax",
            secure=False,     # True in production
        )
    ],
)

CORSMiddleware¶

The default parameters used by the CORSMiddleware implementation are restrictive by default and Lilya allows some ways of using this middleware depending of the taste.

from __future__ import annotations

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.cors import CORSMiddleware

routes = [...]

# Option one
middleware = [DefineMiddleware(CORSMiddleware, allow_origins=["*"])]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(CORSMiddleware, allow_origins=["*"]),
        ]

SessionMiddleware¶

Adds signed cookie-based HTTP sessions. Session information is readable but not modifiable.

from __future__ import annotations

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.sessions import SessionMiddleware
from lilya.requests import Request
from lilya.routing import Path


async def set_session(request: Request) -> dict:
    # this creates/updates the session
    if not request.scope["session"]:
        request.scope["session"] = {"id": 1, "seen_page_list": []}
    else:
        # this updates the session
        request.scope["session"]["seen_page_list"].append("homepage")
    return request.scope["session"]


async def delete_session(request: Request) -> dict:
    old_session = request.scope["session"]
    request.scope["session"] = None
    return old_session

routes = [Path("/set", set_session), Path("/delete", delete_session)]

# Option one
middleware = [DefineMiddleware(SessionMiddleware, secret_key=...)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(SessionMiddleware, secret_key=...),
        ]

By default session data is restricted to json.dumps serializable data. If you want more speed or more datatypes you can pass a different serializer/deserializer, e.g. orjson:

from __future__ import annotations

import orjson
from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.sessions import SessionMiddleware



routes = [...]

# Option one
middleware = [DefineMiddleware(SessionMiddleware, session_serializer=orjson.dumps, session_deserializer=orjson.loads, secret_key=...)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(SessionMiddleware, session_serializer=orjson.dumps, session_deserializer=orjson.loads, secret_key=...),
        ]

Note however when using json not all datatypes are idempotent. You might want to use dataclasses, msgstruct, RDF or (not recommended because of security issues: pickle).

You can also automatically initialize sessions with data via the populate_session parameter. It isn't repopulated until the next request after the session is cleared. So you can change data as you wish.

from __future__ import annotations

from dataclasses import dataclass

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.sessions import SessionMiddleware
from lilya.requests import Request, Connection
from lilya.routing import Path


async def populate_session(con: Connection) -> dict:
    return {"id": 1, "seen_page_list": []}


async def update_session(request: Request) -> dict:
    # this updates the session
    request.scope["session"]["seen_page_list"].append("homepage")
    return request.scope["session"]


async def delete_session(request: Request) -> dict:
    old_session = request.scope["session"]
    request.scope["session"] = None
    return old_session

routes = [Path("/update", update_session), Path("/delete", delete_session)]

middleware = [DefineMiddleware(SessionMiddleware, secret_key=..., populate_session=populate_session)]

app = Lilya(routes=routes, middleware=middleware)

HTTPSRedirectMiddleware¶

Enforces that all incoming requests must either be https or wss. Any http os ws will be redirected to the secure schemes instead.

from __future__ import annotations

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.httpsredirect import HTTPSRedirectMiddleware

routes = [...]

# Option one
middleware = [DefineMiddleware(HTTPSRedirectMiddleware)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(HTTPSRedirectMiddleware),
        ]

TrustedHostMiddleware¶

Enforces all requests to have a correct set Host header in order to protect against host header attacks.

More details in TrustedHostMiddleware

TrustedReferrerMiddleware¶

Check host and referer header to check if the referral was allowed and set the variable in the scope.

More details in TrustedReferrerMiddleware

GZipMiddleware¶

It handles GZip responses for any request that includes "gzip" in the Accept-Encoding header.

from __future__ import annotations

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.compression import GZipMiddleware

routes = [...]

middleware = [DefineMiddleware(GZipMiddleware, minimum_size=1000)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(GZipMiddleware, minimum_size=1000),
        ]

WSGIMiddleware¶

A middleware class in charge of converting a WSGI application into an ASGI one. There are some more examples in the WSGI Frameworks section.

from flask import Flask, make_response

from lilya.apps import Lilya
from lilya.middleware.wsgi import WSGIMiddleware
from lilya.routing import Include

flask = Flask(__name__)


@flask.route("/home")
def home():
    return make_response({"message": "Serving via flask"})


# Add the flask app into Lilya to be served by Lilya.
routes = [Include("/external", app=WSGIMiddleware(flask))]

app = Lilya(routes=routes)

The WSGIMiddleware also allows to pass the app as a string <dotted>.<path> and this can make it easier for code organisation.

Let us assume the previous example of the flask app was inside myapp/asgi_or_wsgi/apps. Like this:

from flask import Flask, make_response

flask = Flask(__name__)


@flask.route("/home")
def home():
    return make_response({"message": "Serving via flask"})

To call it inside the middleware is as simple as:

from lilya.apps import Lilya
from lilya.middleware.wsgi import WSGIMiddleware
from lilya.routing import Include

# Add the flask app into Lilya to be served by Lilya.
routes = [
    Include(
        "/external",
        app=WSGIMiddleware("myapp.asgi_or_wsgi.apps.flask"),
    ),
]

app = Lilya(routes=routes)

XFrameOptionsMiddleware¶

The clickjacking middleware that provides easy-to-use protection against clickjacking. This type of attack occurs when a malicious site tricks a user into clicking on a concealed element of another site which they have loaded in a hidden frame or iframe.

This middleware reads the value x_frame_options from the settings and defaults to DENY.

This also adds the X-Frame-Options to the response headers.

from __future__ import annotations

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.clickjacking import XFrameOptionsMiddleware

routes = [...]

# Option one
middleware = [DefineMiddleware(XFrameOptionsMiddleware)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
class AppSettings(Settings):
    x_frame_options: str = "SAMEORIGIN"

    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(XFrameOptionsMiddleware),
        ]

SecurityMiddleware¶

Provides several security enhancements to the request/response cycle and adds security headers to the response.

from __future__ import annotations

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.security import SecurityMiddleware

routes = [...]

content_policy_dict = {
    "default-src": "'self'",
    "img-src": [
        "*",
        "data:",
    ],
    "connect-src": "'self'",
    "script-src": "'self'",
    "style-src": ["'self'", "'unsafe-inline'"],
    "script-src-elem": [
        "https://unpkg.com/@stoplight/elements/web-components.min.jss",
    ],
    "style-src-elem": [
        "https://unpkg.com/@stoplight/elements/styles.min.css",
    ],
}

# Option one
middleware = [DefineMiddleware(SecurityMiddleware, content_policy=content_policy_dict)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
class AppSettings(Settings):

    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(SecurityMiddleware, content_policy=content_policy_dict),
        ]

ClientIPMiddleware & ClientIPScopeOnlyMiddleware¶

Parses the client ip and add it in the request scope as real-clientip entry. It also adds the standard x-real-ip to the request headers.

from __future__ import annotations

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.clientip import ClientIPMiddleware
from lilya.routing import Path
from lilya.responses import Response
from lilya.requests import Request

async def echo_ip_scope(request: Request) -> Response:
    return Response(f"Your ip is: {request.scope['real-clientip']}")

async def echo_ip_header(request: Request) -> Response:
    return Response(f"Your ip is: {request.headers['x-real-ip']}")

routes = [Path("/echo_scope", handler=echo_ip_scope), Path("/echo_header", handler=echo_ip_header)]

# Only trust unix (no client ip is set) (default)
trusted_proxies = ["unix"]
# No forwarded ip headers will be evaluated. Only the direct ip is accepted
trusted_proxies = []
# trust all client ips to provide forwarded headers
trusted_proxies = ["*"]


# Option one
middleware = [DefineMiddleware(ClientIPMiddleware, trusted_proxies=trusted_proxies)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
class AppSettings(Settings):

    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(ClientIPMiddleware, trusted_proxies=trusted_proxies),
        ]

There are two special "ip"s: "*" and "unix".

The first one is a match all and implies all proxies are trustworthy, the second one applies when a unix socket is used or no client ip address was found.

If you don't want an injected header use the ClientIPScopeOnlyMiddleware:

from __future__ import annotations

from lilya.apps import Lilya
from lilya.middleware import DefineMiddleware
from lilya.middleware.clientip import ClientIPScopeOnlyMiddleware
from lilya.routing import Path
from lilya.responses import Response
from lilya.requests import Request

async def echo_ip_scope(request: Request) -> Response:
    assert "x-real-ip" not in request.headers
    return Response(f"Your ip is: {request.scope['real-clientip']}")


routes = [Path("/echo", handler=echo_ip_scope)]

# Only trust unix (no client ip is set) (default)
trusted_proxies = ["unix"]
# No forwarded ip headers will be evaluated. Only the direct ip is accepted
trusted_proxies = []
# trust all client ips to provide forwarded headers
trusted_proxies = ["*"]


middleware = [DefineMiddleware(ClientIPScopeOnlyMiddleware, trusted_proxies=trusted_proxies)]

app = Lilya(routes=routes, middleware=middleware)

Note

If you don't want to use the middleware you can use: get_ip from lilya.clientip directly.

Note

It is currently not possible to simulate a client ip address in lilyas TestClient. So you may want to use the Forwarded header and trust "unix" for tests.

SessionFixingMiddleware¶

Sometimes you want to tie a session to an ip. If the IP changes, the session is resetted and optionally a notification is sent. This way we can prevent session stealing / session replay attacks. It requires the SessionMiddleware and the ClientIPMiddleware (or the ClientIPScopeOnlyMiddleware).

from __future__ import annotations

from lilya.apps import Lilya
from lilya.middleware import DefineMiddleware
from lilya.middleware.clientip import ClientIPScopeOnlyMiddleware
from lilya.middleware.sessions import SessionMiddleware
from lilya.middleware.session_fixing import SessionFixingMiddleware
from lilya.routing import Path
from lilya.responses import Response
from lilya.requests import Request
from typing import Any

async def echo_ip_scope(request: Request) -> Response:
    assert "x-real-ip" not in request.headers
    return Response(f"Your ip is: {request.scope['real-clientip']}")


routes = [Path("/echo", handler=echo_ip_scope)]

# Only trust unix (no client ip is set) (default)
trusted_proxies = ["unix"]

def notify_fn(old_ip: str | None, new_ip: str, old_session: dict, new_session: dict) -> None:
    if old_ip is None:
        print(f'New session for ip: "{new_ip}".')
    else:
        print(f'Replace session for ip: "{old_ip}". Has new ip "{new_ip}".')

middleware = [
    DefineMiddleware(ClientIPScopeOnlyMiddleware, trusted_proxies=trusted_proxies),
    DefineMiddleware(SessionMiddleware, secret_key=...),
    DefineMiddleware(SessionFixingMiddleware,notify_fn=notify_fn)
]

app = Lilya(routes=routes, middleware=middleware)

Note

Drawback is a periodically logout if you have a client with a changing ip address. E.g. working remotely from trains or using tor.

Tip

You can use the notify_fn parameter to update the new_session with data from the old_session when appropiate.

RequestContextMiddleware¶

Lazy loads a request object without explicitly add it into the handlers.

from __future__ import annotations

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.request_context import RequestContextMiddleware

routes = [...]

# Option one
middleware = [DefineMiddleware(RequestContextMiddleware)]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(RequestContextMiddleware),
        ]

SessionContextMiddleware¶

Lazy loads a session object without explicitly go through the request.session object. This can be accessed within the request context of any handler.

The SessionContextMiddleware depends on the SessionMiddleware to also be installed and the order matters.

from __future__ import annotations

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.middleware import DefineMiddleware
from lilya.middleware.session_context import SessionContextMiddleware
from lilya.middleware.sessions import SessionMiddleware

routes = [...]

# Option one
middleware = [
    DefineMiddleware(SessionMiddleware, secret="my_secret"),
    DefineMiddleware(SessionContextMiddleware)
]

app = Lilya(routes=routes, middleware=middleware)


# Option two - Using the settings module
# Running the application with your custom settings -> LILYA_SETTINGS_MODULE
class AppSettings(Settings):
    @property
    def middleware(self) -> list[DefineMiddleware]:
        return [
            DefineMiddleware(SessionMiddleware, secret='my_secret'),
            DefineMiddleware(SessionContextMiddleware),
        ]

You can multiplex a single session by providing sub_path:

from __future__ import annotations

from dataclasses import dataclass

from lilya.apps import Lilya
from lilya.routing import Include
from lilya.middleware import DefineMiddleware
from lilya.middleware.session_context import SessionContextMiddleware
from lilya.middleware.sessions import SessionMiddleware

routes = [
    Include(
        "",
        routes=[],
        middleware=[
            DefineMiddleware(SessionContextMiddleware, sub_path="path1")
        ]
    ),

    Include(
        "",
        routes=[],
        middleware=[
            DefineMiddleware(SessionContextMiddleware, sub_path="path2")
        ]
    )
]

middleware = [
    DefineMiddleware(SessionMiddleware, secret="my_secret")
]

app = Lilya(routes=routes, middleware=middleware)

Note

This doesn't affect request.session or scope["session"]. Here you have still the original session object.

Other middlewares¶

You can build your own middlewares as explained above but also reuse middlewares directly for any other ASGI application if you wish. If the middlewares follow the pure asgi then the middlewares are 100% compatible.

RateLimitMiddleware ¶

A ASGI Middleware to rate limit and highly customizable.

CorrelationIdMiddleware ¶

A middleware class for reading/generating request IDs and attaching them to application logs.

Tip

For Lilya apps, just substitute FastAPI with Lilya in the examples given or implement in the way Lilya shows in this document.

TimingMiddleware ¶

ASGI middleware to record and emit timing metrics (to something like statsd).

Important points¶

Lilya supports Lilya middleware (MiddlewareProtocol).
A MiddlewareProtocol is simply an interface that enforces __init__ and async __call__ to be implemented.
app is required parameter from any class inheriting from the MiddlewareProtocol.
Pure ASGI Middleware is encouraged and the MiddlewareProtocol enforces that.
Middleware classes can be added to any layer of the application
All authentication middlewares must inherit from the BaseAuthMiddleware.
You can load the application middleware in different ways.

Middleware¶

Lilya middleware¶

Lilya protocols¶

MiddlewareProtocol¶

Quick sample¶

Middleware and the application¶

Quick note¶

Pure ASGI Middleware¶

BaseAuthMiddleware & AuthenticationMiddleware¶

Example of a JWT middleware class¶

Import the middleware into a Lilya application¶

Middleware and the settings¶

Important¶

Available middlewares¶

CSRFMiddleware¶

What it protects and how¶

Behaviour at a glance¶

Examples¶

1. Classic HTML form (no JavaScript)¶

2. XHR / fetch (header path)¶

3. File uploads (multipart)¶

4. Custom form field name¶

5. Tightening body buffering¶

Minimal, complete examples¶

CORSMiddleware¶

SessionMiddleware¶

HTTPSRedirectMiddleware¶

TrustedHostMiddleware¶

TrustedReferrerMiddleware¶

GZipMiddleware¶

WSGIMiddleware¶

XFrameOptionsMiddleware¶

SecurityMiddleware¶

ClientIPMiddleware & ClientIPScopeOnlyMiddleware¶

SessionFixingMiddleware¶

RequestContextMiddleware¶

SessionContextMiddleware¶

Other middlewares¶

RateLimitMiddleware¶

CorrelationIdMiddleware¶

TimingMiddleware¶

Important points¶

RateLimitMiddleware ¶

CorrelationIdMiddleware ¶

TimingMiddleware ¶