JWT Implementation done

Author: Jrhero14

routes/api.py

@@ -1,10 +1,13 @@
+from typing import Annotated
+
 from starlette.responses import JSONResponse
 
 from app.controllers import authController
-from fastapi import APIRouter, status
+from fastapi import APIRouter, status, Depends
 
 from app.dto.CreateUserDto import CreateUserDto
 from app.dto.LoginUserDto import LoginUserDto
+from utilities.jwtUtils import validate_token
 
 router = APIRouter(
     prefix="/api",
@@ -24,4 +27,10 @@ async def registerUser(bodyRequest: CreateUserDto):
 @router.post("/login", status_code=status.HTTP_200_OK, response_class=JSONResponse)
 async def login(bodyRequest: LoginUserDto):
     response = await authController.login(body=bodyRequest)
-    return response
+    return response
+
+@router.post("/validate-token", status_code=status.HTTP_200_OK, response_class=JSONResponse, dependencies=[Depends(validate_token)])
+async def login():
+    return {
+        "message": "Token valid"
+    }

utilities/jwtUtils.py

@@ -1,10 +1,10 @@
 from datetime import datetime, timedelta, timezone
-from typing import Annotated, Union
+from typing import Annotated, Union, Any
 
 import jwt, os
 from dotenv import load_dotenv
 from fastapi import Depends, FastAPI, HTTPException, status
-from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from jwt.exceptions import InvalidTokenError
 from passlib.context import CryptContext
 from pydantic import BaseModel
@@ -15,7 +15,7 @@
 ALGORITHM = os.getenv("HASH_ALGORITHM")
 ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+oauth2_scheme = HTTPBearer(scheme_name="JWT")
 
 class TokenData(BaseModel):
     username: Union[str, None] = None
@@ -33,21 +33,15 @@ def create_access_token(payload: dict, expires_delta: Union[timedelta, None] = N
     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
     return encoded_jwt
 
-# async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
-#     credentials_exception = HTTPException(
-#         status_code=status.HTTP_401_UNAUTHORIZED,
-#         detail="Could not validate credentials",
-#         headers={"WWW-Authenticate": "Bearer"},
-#     )
-#     try:
-#         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
-#         username: str = payload.get("sub")
-#         if username is None:
-#             raise credentials_exception
-#         token_data = TokenData(username=username)
-#     except InvalidTokenError:
-#         raise credentials_exception
-#     user = get_user(fake_users_db, username=token_data.username)
-#     if user is None:
-#         raise credentials_exception
-#     return user
+def validate_token(token: Annotated[HTTPAuthorizationCredentials, Depends(oauth2_scheme)]):
+    try:
+        payload = jwt.decode(token.credentials, SECRET_KEY, algorithms=[ALGORITHM])
+        user_id: int = payload.get("user_id")
+        if user_id is None:
+            raise InvalidTokenError
+    except InvalidTokenError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token Invalid or Expired",
+            headers={"WWW-Authenticate": "Bearer"},
+        )