How to inspect every request (including request body) with fastapi?

Question

I would like to create a middleware that authorizes every request based on its url, headers and body (OPA). To do that I created a custom request and custom route class. It is the advised way if one wants to manipulate the request body. Firstly, I tried to override the .json() method and do the authorization there, however, not every request handler calls the .json() method, so that is not a possible solution. Then I decided to create a custom method on the request that will do the authorization and call it from a custom middleware. The issue is that the middleware receives a plain request, which does not have the .is_authorized method, rather than the subclassed request.

Minimal example:

import json
from typing import Callable, Any

from fastapi import FastAPI, Request, Response
from fastapi.routing import APIRoute
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel


def some_authz_func(request: Request, body: Any) -> bool:
    print(vars(request), body)
    return True


class MessageModel(BaseModel):
    message: str


class AuthzMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        await request.is_authorized()
        response = await call_next(request)
        return response


class AuthzRequest(Request):
    async def is_authorized(self) -> bool:
        try:
            json_ = await super().json()
        except json.decoder.JSONDecodeError:
            json_ = None
        return some_authz_func(self, json_)


class AuthzRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            request = AuthzRequest(request.scope, request.receive)
            return await original_route_handler(request)

        return custom_route_handler


app = FastAPI(route_class=AuthzRoute, middleware=[Middleware(AuthzMiddleware)])


@app.get('/')
async def no_body():
    return JSONResponse(content={'no-body': True})


@app.post('/')
async def some_body(body: MessageModel):
    print(body)
    return JSONResponse(content={'body': True})

Raised exception

  File ".virtualenvs/tempenv-416e126962124/lib/python3.8/site-packages/starlette/middleware/errors.py", line 162, in __call__
    await self.app(scope, receive, _send)
  File ".virtualenvs/tempenv-416e126962124/lib/python3.8/site-packages/starlette/middleware/base.py", line 68, in __call__
    response = await self.dispatch_func(request, call_next)
  File "main.py", line 23, in dispatch
    await request.is_authorized()
AttributeError: 'Request' object has no attribute 'is_authorized'

Dependencies

$ pip freeze
anyio==3.6.1
click==8.1.3
fastapi==0.79.0
h11==0.13.0
idna==3.3
pydantic==1.9.1
sniffio==1.2.0
starlette==0.19.1
typing_extensions==4.3.0
uvicorn==0.18.2

Answer 1

It can be solved by using dependency injection and applying it to the app object (Thanks @MatsLindh). The dependency function can take a Request object and get the ulr, headers and body from it.

from fastapi import FastAPI, Request, Depends

async def some_authz_func(request: Request):
    try:
        json_ = await request.json()
    except json.decoder.JSONDecodeError:
        json_ = None
    print(vars(request), json_)


app = FastAPI(dependencies=[Depends(some_authz_func)])

Whole working example

import json

from fastapi import FastAPI, Request, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel


async def some_authz_func(request: Request):
    try:
        json_ = await request.json()
    except json.decoder.JSONDecodeError:
        json_ = None
    print(vars(request), json_)


app = FastAPI(dependencies=[Depends(some_authz_func)])


class MessageModel(BaseModel):
    message: str


@app.get('/')
async def no_body():
    return JSONResponse(content={'no-body': True})


@app.post('/')
async def some_body(body: MessageModel):
    print(body)
    return JSONResponse(content={'body': True})