14

I am trying to validate token in fastapi middleware but this seems impossible. As i am thinking middleware needs to make next call although its not required. I am not able to find any good solution to handle token in one go in this python-fastapi backend. Any help is appreciated.

@app.middleware("http")
async def add_middleware_here(request: Request, call_next):
    token = request.headers["Authorization"]
    try:
        verification_of_token = verify_token(token)
        if verification_of_token:
            response = await call_next(request)
            return response
    except InvalidSignatureError as er:
        raise HTTPException(status_code=401)
TrickOrTreat
  • 821
  • 1
  • 9
  • 23
  • I don't find any mention that that it is possible. Though you may try responding with a custom response https://fastapi.tiangolo.com/advanced/response-directly/ – lsabi Apr 23 '20 at 13:36
  • Did you find a solution? I have the same issue. @TrickOrTreat – Saeed Esmaili May 29 '20 at 12:06

2 Answers2

14

You need to return a response. I'll show you how you can make it work:

from fastapi.responses import JSONResponse

@app.middleware("http")
async def add_middleware_here(request: Request, call_next):
    token = request.headers["Authorization"]
    try:
        verification_of_token = verify_token(token)
        if verification_of_token:
            response = await call_next(request)
            return response
        else:
            return JSONResponse(status_code=403) # or 401
    except InvalidSignatureError as er:
        return JSONResponse(status_code=401)

Be aware that using this middleware implies that you don't have a login route on your API (for token generation).

Besides, you should consider using this dependency instead: https://fastapi.tiangolo.com/tutorial/security/simple-oauth2/

Another thing is that you can use fastapi.status for your status code.

Marcelo Trylesinski
  • 634
  • 1
  • 5
  • 13
3

See JSONResponse API documentation.

from fastapi.responses import JSONResponse

@app.middleware("http")
async def errors_handling(request: Request, call_next):
    try:
        return await call_next(request)
    except Exception as exc:
        return JSONResponse(status_code=500, content={'reason': str(exc)})