Authentication Middleware

This special middleware comes from Lilya and is used to authenticate users on every request but adapted for Ravyn.

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

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

Majority of the things here are from Lilya and we just added the ability to use it as a Ravyn middleware with almost no changes.

Example of a JWT middleware class

/src/middleware/jwt.py

from typing import Sequence
from myapp.models import User
from myapp.security.jwt.token import Token
from edgy.exceptions import ObjectNotFound

from lilya._internal._connection import Connection
from lilya.authentication import AuthResult, AuthCredentials
from lilya.types import ASGIApp

from ravyn.exceptions import NotAuthorized
from ravyn.middleware.authentication import AuthenticationMiddleware, AuthenticationBackend


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

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

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

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

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


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

    def __init__(
        self,
        app: ASGIApp,
        signing_key: str,
        algorithm: str,
        api_key_header: str,
        backend: Sequence[AuthenticationBackend] | AuthenticationBackend | None = None,
    ):
        super().__init__(app, backend=backend)
        self.app = app
        self.signing_key = signing_key
        self.algorithm = algorithm
        self.api_key_header = api_key_header

    async def authenticate(self, conn: Connection) -> AuthResult | Exception:
        for backend in self.backend:
            # exceptions are passed through to __call__ and there handled
            auth_result = await backend.authenticate(
                conn,
                signing_key=self.signing_key,
                algorithm=self.algorithm,
                api_key_header=self.api_key_header,
            )
            if auth_result is not None:
                return auth_result
        return NotAuthorized()

1. Import the AuthenticationMiddleware from ravyn.middleware.authentication.
2. 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 ravyn import Ravyn
from lilya.middleware import DefineMiddleware
from .middleware.jwt import JWTAuthMiddleware


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

from ravyn import Ravyn, RavynSettings
from lilya.middleware import DefineMiddleware


class AppSettings(RavynSettings):

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

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

Tip

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

Authentication

Ravyn provides a straightforward yet robust interface for managing authentication and permissions (from Lilya). By installing AuthenticationMiddleware with a suitable authentication backend, you can access the request.user and request.auth interfaces within your endpoints.

import base64
import binascii

from lilya.authentication import AuthCredentials, AuthenticationBackend, BasicUser
from lilya.middleware import DefineMiddleware

from ravyn import Ravyn, Gateway, Request, get
from ravyn.exceptions import AuthenticationError
from ravyn.middleware.sessions import SessionMiddleware
from ravyn.middleware.authentication import AuthenticationMiddleware
from ravyn.responses import PlainText


class SessionBackend(AuthenticationBackend):
    async def authenticate(self, connection):
        if "session" not in connection.scope:
            return

        if connection.scope["session"].get("username", None):
            return
        return AuthCredentials(["authenticated"]), BasicUser(
            connection.scope["session"]["username"]
        )


class BasicAuthBackend(AuthenticationBackend):
    async def authenticate(self, connection):
        if "Authorization" not in connection.headers:
            return

        auth = connection.headers["Authorization"]
        try:
            scheme, credentials = auth.split()
            if scheme.lower() != "basic":
                return
            decoded = base64.b64decode(credentials).decode("ascii")
        except (ValueError, UnicodeDecodeError, binascii.Error) as exc:
            raise AuthenticationError("Invalid basic auth credentials")

        username, _, password = decoded.partition(":")
        return AuthCredentials(["authenticated"]), BasicUser(username)


@get()
async def homepage(request: Request) -> PlainText:
    if request.user.is_authenticated:
        return PlainText("Hello, " + request.user.display_name)
    return PlainText("Hello, you")


app = Ravyn(
    routes=[Gateway("/", handler=homepage)],
    middleware=[
        # must be defined before AuthenticationMiddleware, because of the SessionBackend
        DefineMiddleware(SessionMiddleware, secret_key=...),
        DefineMiddleware(AuthenticationMiddleware, backend=[SessionBackend(), BasicAuthBackend()]),
    ],
)

When not using an user management we can also do something like:

import base64
import binascii

from lilya.authentication import AuthCredentials, AuthenticationBackend, BasicUser
from lilya.middleware import DefineMiddleware

from ravyn import Ravyn, Gateway, Request, get
from ravyn.exceptions import AuthenticationError
from ravyn.middleware.authentication import AuthenticationMiddleware
from ravyn.responses import PlainText
import secrets


class HardCodedBasicAuthBackend(AuthenticationBackend):
    def __init__(self, *, username: str = "admin", password: str) -> None:
        self.basic_string = base64.b64encode(f"{username}:{password}".encode()).decode()

    async def authenticate(self, connection) -> tuple[AuthCredentials, BasicUser] | None:
        if "Authorization" not in connection.headers:
            return None

        auth = connection.headers["Authorization"]
        try:
            scheme, credentials = auth.split()
            if scheme.lower() != "basic":
                return None
            if not secrets.compare_digest(credentials, self.basic_string):
                raise ValueError()
            username = base64.b64decode(credentials).decode("ascii").split(":", 1)[0]
        except (ValueError, UnicodeDecodeError, binascii.Error) as exc:
            raise AuthenticationError("Invalid basic auth credentials")

        return AuthCredentials(["authenticated"]), BasicUser(username)


@get()
async def homepage(request: Request) -> PlainText:
    if request.user.is_authenticated:
        return PlainText("Hello, " + request.user.display_name)
    return PlainText("Hello, you")


app = Ravyn(
    routes=[Gateway("/", handler=homepage)],
    middleware=[
        DefineMiddleware(
            AuthenticationMiddleware, backend=[HardCodedBasicAuthBackend(password="password")]
        )
    ],
)

authenticate() from AuthenticationMiddleware

This is the main method that goes through the backends and tries to authenticate the user.

Sometimes you want to override this method to add some custom logic before or after the authentication.

project/middleware/jwt.py

from typing import Sequence
from myapp.models import User
from myapp.security.jwt.token import Token
from edgy.exceptions import ObjectNotFound

from lilya._internal._connection import Connection
from lilya.authentication import AuthResult, AuthCredentials
from lilya.types import ASGIApp

from ravyn.exceptions import NotAuthorized
from ravyn.middleware.authentication import AuthenticationMiddleware, AuthenticationBackend


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

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

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

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

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


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

    def __init__(
        self,
        app: ASGIApp,
        signing_key: str,
        algorithm: str,
        api_key_header: str,
        backend: Sequence[AuthenticationBackend] | AuthenticationBackend | None = None,
    ):
        super().__init__(app, backend=backend)
        self.app = app
        self.signing_key = signing_key
        self.algorithm = algorithm
        self.api_key_header = api_key_header

    async def authenticate(self, conn: Connection) -> AuthResult | Exception:
        for backend in self.backend:
            # exceptions are passed through to __call__ and there handled
            auth_result = await backend.authenticate(
                conn,
                signing_key=self.signing_key,
                algorithm=self.algorithm,
                api_key_header=self.api_key_header,
            )
            if auth_result is not None:
                return auth_result
        return NotAuthorized()

Here we passed extra information to the authenticate of the backend to be used for internal operations.

Backends

For backends you need the AuthenticationMiddleware (not the BaseAuthMiddleware from Lilya). Only here you can provide them via the backend parameter. This can be a sequence of AuthenticationBackend instances or also a single one.

If a backend doesn't find the user it can return None in authenticate to skip to the next Backend.

If a backend raises an error in authenticate, the whole chain is stopped.

Backends are retrievable on the middleware via the backend attribute. It is always a list.

Users

Once you have installed AuthenticationMiddleware, the request.user interface becomes available to your endpoints and other middleware.

The implementation should implement the interface UserInterface, which includes two properties and any additional information your user model requires.

• .is_authenticated
• .display_name

Ravyn provides two built-in user implementations: AnonymousUser() and BasicUser(username).

AuthCredentials

Authentication credentials should be considered distinct from user identities. An authentication scheme must be capable of granting or restricting specific privileges independently of the user's identity.

The AuthCredentials class provides the basic interface that request.auth exposes:

• .scopes

Custom authentication error responses

You can customize the error response sent when an AuthenticationError is raised by an authentication backend:

from lilya.middleware import DefineMiddleware

from ravyn import JSONResponse, Ravyn, Request
from ravyn.middleware.authentication import AuthenticationMiddleware


def on_auth_error(request: Request, exc: Exception):
    return JSONResponse({"error": str(exc)}, status_code=401)

app = Ravyn(
    middleware=[
        DefineMiddleware(AuthenticationMiddleware, backend=BasicAuthBackend(), on_error=on_auth_error),
    ],
)