(HELP) FastAPI get current user - JWT Token

[naheedroomy]

So I am currently writing a User management API with FastAPI and certain endpoints are protected by this JWT Token. The JWT authorization is written as follows. I did not use the oauth2 method prescribed by FastAPI for certain reasons.

from fastapi import Request, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

from auth.auth_handler import decode_jwt


def verify_jwt(token_jwt: str) -> bool:
    is_token_valid: bool = False

    try:
        payload = decode_jwt(token_jwt)
    except Exception:
        payload = None
    if payload:
        is_token_valid = True
    return is_token_valid


class JWTBearer(HTTPBearer):
    def __init__(self, auto_error: bool = True):
        super(JWTBearer, self).__init__(auto_error=auto_error)

    async def __call__(self, request: Request):
        credentials: HTTPAuthorizationCredentials = await super(
            JWTBearer,
            self).__call__(request)
        if credentials:
            if not credentials.scheme == "Bearer":
                raise HTTPException(status_code=403,
                                    detail="Invalid authentication scheme.")
            if not verify_jwt(credentials.credentials):
                raise HTTPException(status_code=403,
                                    detail="Invalid token or expired token.")
            return credentials.credentials
        else:
            raise HTTPException(status_code=403,
                                detail="Invalid authorization code.")

import hashlib
import jwt
import time
from decouple import config
from models.user import MongoUser
from typing import Dict

JWT_SECRET = config("secret")
JWT_ALGORITHM = config("algorithm")


def token_response(token: str):
    return {
        "access_token": token
    }


def sign_jwt(user_id: str) -> Dict[str, str]:
    payload = {
        "user_id": user_id,
        "expires": time.time() + 600
    }
    token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)

    return token_response(token)


def decode_jwt(token: str) -> dict:
    try:
        decoded_token = jwt.decode(token,
                                   JWT_SECRET,
                                   algorithms=[JWT_ALGORITHM])
        return decoded_token if decoded_token["expires"] >= \
                                time.time() else None
    except Exception:
        return {}


def authenticate_user(email, password):
    hashed = hashlib.sha256(password.encode('utf-8'))
    hashed_password = hashed.hexdigest()
    verification = MongoUser.objects.filter(
        email=email,
        password=hashed_password)
    if verification:
        return verification

A sample endpoint that depends on the above authorization would be 
@user.post('/profile',
           dependencies=[Depends(JWTBearer())],
           responses={404: {"model": Message}},
           tags=["User Profile"])
async def create_user_profile(user_profile: UserProfile,
                              conn=Depends(connect_db())):
    api = "[USER PROFILE]"
    selected_user = MongoUser.objects(email=user_profile.email)
    userid = str(selected_user[0].id)
    profile_exist = MongoUserProfile.objects.filter(_id=userid)
    if not profile_exist:

        MongoUserProfile(_id=userid,
                         role=user_profile.role,
                         focus_area=user_profile.focus_area,
                         projects=user_profile.projects).save()
        logger.info(api + ": Created DB entry")
        return JSONResponse(
            status_code=200,
            content={"message": "User profile created successfully"}
        )
    else:
        return JSONResponse(
            status_code=403,
            content={"message": "User profile already exist"}
        )

I want to be able to extract the user email from the token, Is this possible? I did see an implementation of this in the FastAPI documentation - https://fastapi.tiangolo.com/tutorial/security/get-current-user/ however, me being new to FastAPI and stuff, I just cannot figure out how to implement it in my project.

Can someone guide me through this?

TIA!

[negadive]

you could make dependency that take jwt and decode to get the claims/payload then return it

async def get_current_user(token: str = Depends(JWTBearer())) -> dict:
    # skipping verify since its already verified in JWTBearer
    payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM], verify_signature=False) 
    return {
        "email": payload.get("email")
    }

you also could save the payloads in request.state while verifying in JWTBearer to skip redecoding the token

[naheedroomy]

Thank you! Was able to solve it after this. <3

[zobayer1]

Is there a way to do this without using Depends()?