Permissions decorator

[florimondmanca]

Problem

Starlette provides an endpoint decorator to check for permissions. However, this endpoint is tied to Starlette's view system and may not be reusable in all situations.

Solution you'd like

To free ourselves from the constraint on how views are implemented, we need to use ASGI. Assuming that routes are ASGI applications themselves (this is a potential blocker, but most ASGI frameworks work like this anyway), we could write an ASGI route decorator as follows:

import typing

from starlette.requests import Request, HTTPConnection
from starlette.websockets import WebSocket
from starlette.responses import RedirectResponse
from starlette.exceptions import HTTPException


class PermissionsDecorator:
    def __init__(self, status_code: int = 403):
        self.status_code = status_code

    def has_required_scope(self, conn: HTTPConnection, scopes: typing.Sequence[str]) -> bool:
        return all(scope in conn.auth.scopes for scope in scopes)

    # Overridable methods with Starlette-specific defaults

    async def redirect(self, scope, receive, send, url: str):
        response = RedirectResponse(url)
        await response(scope, receive, send)

    def http_error(self, status_code: int):
        raise HTTPException(status_code=status_code)

    # Decorator implementation

    def __call__(*scopes: str, status_code: int = None, redirect: str = None):
        if status_code is None:
            status_code = self.status_code

        def decorate(route):
            async def wrapped_route(scope, receive, send):
                assert scope["type"] in {"http", "websocket"}

                if scope["type"] == "http":
                    request = Request(scope, receive)
                    if not self.has_required_scope(request, scopes):
                        if redirect is not None:
                            await self.redirect(scope, receive, send, url=redirect)
                            return
                        self.http_error(status_code=status_code)
               else:
                   websocket = WebSocket(scope, receive, send)
                   if not self.has_required_scope(websocket, scopes):
                       await websocket.close()
                       return

                await route(scope, receive, send)

            return wrapped_view
        return view


# Export Starlette-specific decorator as default
requires = PermissionsDecorator()

Usage with Starlette:

from starlette.applications import Starlette
from starlette_auth_toolkit.permissions import requires

@requires("authenticated", status_code=404)
@app.route("/secret")
async def secret_view(request):
    return PlainTextResponse("Authorized")