5

I was wondering if it was possible to pass the results from the dependencies kwarg in include_router to the router that is passed to it. What I want to do is decode a JWT from the x-token header of a request and pass the decoded payload to the books routes.

I know that I could just write authenticate_and_decode_JWT as a dependency of each of the routes in routers/book.py, but this would be quite repetitive for a large app.

main.py

from typing import Optional
from jose import jwt

from fastapi import FastAPI, Depends, Header, HTTPException, status
from jose.exceptions import JWTError

from routers import books


app = FastAPI()

def authenticate_and_decode_JWT(x_token: str = Header(None)):
    try:
        payload = jwt.decode(x_token.split(' ')[1], 'secret key', algorithms=['HS256'])
        return payload # pass decoded user information from here to books.router routes somehow
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

app.include_router(
    books.router,
    prefix="/books",
    dependencies=[Depends(authenticate_and_decode_JWT)], 
)

@app.get("/")
def read_root():
    return {"Hello": "World"}

routers/books.py

from fastapi import APIRouter

router = APIRouter()

@router.get('/')
def get_book():
    # do stuff with decoded authenticated user data from JWT payload here
    pass

@router.post('/')
def create_book():
    # do stuff with decoded authenticated user data from JWT payload here
    pass
neilZon
  • 113
  • 3
  • 7

2 Answers2

10

For the larger application you mentioned, You can do following things,

  1. Write a Authentication Backend. So, you can access request.user from every route
class BearerTokenAuthBackend(AuthenticationBackend):
    """
    This is a custom auth backend class which will allow you to authenticate your request and return auth and user as
    a tuple
    """
    async def authenticate(self, request):
        # This function is inherited from the base class and called by some other class
        if "Authorization" not in request.headers:
            return
        auth = request.headers["Authorization"]
        try:
            scheme, token = auth.split()
            if scheme.lower() != 'bearer':
                return
            decoded = jwt.decode(
                token,
                settings.JWT_SECRET,
                algorithms=[settings.JWT_ALGORITHM],
                options={"verify_aud": False},
            )
        except (ValueError, UnicodeDecodeError, JWTError) as exc:
            raise AuthenticationError('Invalid JWT Token.')
        username: str = decoded.get("sub")
        # QUERY FROM DATABASE/CACHE add to user
        user = None
        if user is None:
            raise AuthenticationError('Invalid JWT Token.')
        return auth, user
  1. Add the Middleware to application startup
app.add_middleware(AuthenticationMiddleware, backend=BearerTokenAuthBackend())
  1. You you can access request.user
@router.get('/')
def get_book(request: Request):
    print(request.user)
    # do stuff with decoded authenticated user data from JWT payload here
    pass

I've written this hack for myself. You can suit like yours.

2

While the answer above about writing a middleware does work, if you don't want a middleware, you just want to use the include_router, then you can take the authenticate_and_decode_JWT method and extend it to write the JWT payload into the request object and then later have your routes read from that out from the request object.

This way you only need to modify a few lines in your code for it to work:

from fastapi import Request  # <== new import

def authenticate_and_decode_JWT(x_token: str = Header(None), request: Request):   # <== request is a new param
    try:
        payload = jwt.decode(x_token.split(' ')[1], 'secret key', algorithms=['HS256'])
        request.token_payload = payload  # type: ignore   # <== store the token payload in request
        return payload # pass decoded user information from here to books.router routes somehow
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

And then read from request:

from fastapi import Request   # <== new import

@router.get('/', request: Request)   # <== request is a new param
def get_book():
    # do stuff with decoded authenticated user data from JWT payload here
    print(request.token_payload)   # <== access your token payload 
Zoltan Fedor
  • 2,004
  • 2
  • 23
  • 40