Merge pull request #6 from Ifechukwu001/feat/jwt-auth

Exposed an API for validating JWT tokens for authentication

Author: CaptainAril

src/api/constants/activity_types.py

@@ -9,6 +9,7 @@ class ActivityTypes(TypedDict):
     RESEND_EMAIL: str
     EMAIL_VALIDATION: str
     USER_LOGIN: str
+    VALIDATE_TOKEN: str
     CHANGE_PASSWORD: str
     REQUEST_RESET_PASSWORD: str
     CONFIRM_RESET_PASSWORD: str
@@ -22,6 +23,7 @@ class ActivityTypes(TypedDict):
     "RESEND_EMAIL": "Resend email validation",
     "EMAIL_VALIDATION": "Email validation",
     "USER_LOGIN": "User login",
+    "VALIDATE_TOKEN": "Validate token",
     "CHANGE_PASSWORD": "Change user password",
     "REQUEST_RESET_PASSWORD": "Request for password reset",
     "CONFIRM_RESET_PASSWORD": "Confirm password reset",

src/api/constants/messages.py

@@ -17,6 +17,8 @@ class AuthMessages(TypedDict):
     NOT_ENABLED: str
     IS_DELETED: str
     SUCCESS: str
+    TOKEN_ERROR: str
+    TOKEN_SUCCESS: str
 
 
 class OtpMessages(TypedDict):
@@ -28,30 +30,11 @@ class OtpMessages(TypedDict):
 
 class UserMessages(TypedDict):
     DOESNT_EXIST: str
-    UPDATED: str
     SANITIZED: str
-    NOT_ALLOWED: str
-    PIN_EXISTS: str
-    PIN_SET: str
     INCORRECT_PASSWORD: str
     PASSWORD_CHANGED: str
 
 
-class AccountMessages(TypedDict):
-    SAVED: str
-    UPDATED: str
-
-
-class NextOfKinMessages(TypedDict):
-    FETCHED: str
-    UPDATED: str
-
-
-class KYCInformationMessages(TypedDict):
-    FETCHED: str
-    UPDATED: str
-
-
 class CommonMessages(TypedDict):
     INTERNAL_SERVER_ERROR: str
     JWT_GENERATED: str
@@ -73,9 +56,6 @@ class Messages(TypedDict):
     AUTH: AuthMessages
     OTP: OtpMessages
     USER: UserMessages
-    ACCOUNT: AccountMessages
-    NOK: NextOfKinMessages
-    KYC: KYCInformationMessages
     COMMON: CommonMessages
     PASSWORD_RESET: PasswordResetMessages
 
@@ -116,6 +96,8 @@ class DynamicMessages(TypedDict):
         "NOT_ENABLED": "User account is disabled. Please contact support",
         "IS_DELETED": "User account has been deleted. Please contact support if you want to restore your account",
         "SUCCESS": "Successfully logged in",
+        "TOKEN_ERROR": "Invalid token",
+        "TOKEN_SUCCESS": "Valid token",
     },
     "OTP": {
         "SEND_SUCCESS": "OTP sent successfully",
@@ -125,26 +107,10 @@ class DynamicMessages(TypedDict):
     },
     "USER": {
         "DOESNT_EXIST": "User doesn't exist",
-        "UPDATED": "User updated successfully",
         "SANITIZED": "User object was sanitized",
-        "PIN_EXISTS": "You already have a transaction PIN on your account!",
-        "PIN_SET": "Transaction PIN set successfully",
-        "NOT_ALLOWED": "Unauthorized request!",
         "INCORRECT_PASSWORD": "Incorrect old password",
         "PASSWORD_CHANGED": "Password changed successfully",
     },
-    "ACCOUNT": {
-        "SAVED": "Withdrawal account details saved succesfully",
-        "UPDATED": "User Withdraw account updated successfully",
-    },
-    "NOK": {
-        "FETCHED": "Next of Kin fetched successfully",
-        "UPDATED": "Next of Kin updated successfully",
-    },
-    "KYC": {
-        "FETCHED": "KYC Information fetched successfully",
-        "UPDATED": "KYC Information updated successfully",
-    },
     "COMMON": {
         "INTERNAL_SERVER_ERROR": "Something went wrong",
         "JWT_GENERATED": "JWT was generated",

src/api/controllers/AuthController.py

@@ -6,6 +6,7 @@
 from src.api.constants.messages import MESSAGES
 from src.api.services.AuthService import AuthService
 from src.api.utils.response_format import error_response, success_response
+from src.api.models.payload.requests.JWT import JWT
 from src.api.models.payload.requests.ResendUserOtp import ResendUserOtp
 from src.api.models.payload.requests.CreateUserRequest import CreateUserRequest
 from src.api.models.payload.requests.AuthenticateUserOtp import AuthenticateUserOtp
@@ -71,6 +72,18 @@ async def login(self, credentials: AuthenticateUserRequest) -> tuple:
             status_code=HTTPStatus.OK,
         )
 
+    async def validate_token(self, credentials: JWT) -> tuple:
+        jwt_details = await self.auth_service.validate_token(credentials)
+        if not jwt_details["is_success"]:
+            return error_response(
+                message=jwt_details["message"], status_code=HTTPStatus.UNAUTHORIZED
+            )
+        return success_response(
+            message=jwt_details["message"],
+            data=jwt_details["data"],
+            status_code=HTTPStatus.OK,
+        )
+
     async def change_password(
         self, id: str, user_data: ChangeUserPasswordRequest
     ) -> tuple:

src/api/models/payload/requests/JWT.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class JWT(BaseModel):
+    token: str

src/api/routes/Auth.py

@@ -5,6 +5,7 @@
 
 from src.utils.svcs import ADepends
 from src.api.controllers.AuthController import AuthController
+from src.api.models.payload.requests.JWT import JWT
 from src.api.models.payload.responses.User import UserResponse, UserLoginResponse
 from src.api.models.payload.requests.ResendUserOtp import ResendUserOtp
 from src.api.models.payload.responses.ErrorResponse import (
@@ -78,6 +79,19 @@ async def login(request: HttpRequest, credentials: AuthenticateUserRequest) -> t
     return await auth_controller.login(credentials)
 
 
+@router.post(
+    "/validate-token",
+    response={
+        HTTPStatus.OK: SuccessResponse,
+        HTTPStatus.UNAUTHORIZED: ErrorResponse,
+        HTTPStatus.INTERNAL_SERVER_ERROR: ServerErrorResponse,
+    },
+)
+async def validate_jwt_token(request: HttpRequest, credentials: JWT) -> tuple:
+    auth_controller = await ADepends(AuthController)
+    return await auth_controller.validate_token(credentials)
+
+
 @router.put(
     "/change-password",
     response={

src/api/services/AuthService.py

@@ -2,10 +2,12 @@
 
 from src.utils.svcs import Service
 from src.utils.logger import Logger
+from src.api.typing.JWT import JWTSuccess
 from src.api.typing.UserExists import UserExists
 from src.api.constants.messages import MESSAGES, DYNAMIC_MESSAGES
 from src.api.typing.UserSuccess import UserSuccess
 from src.api.constants.activity_types import ACTIVITY_TYPES
+from src.api.models.payload.requests.JWT import JWT
 from src.api.repositories.UserRepository import UserRepository
 from src.api.models.payload.requests.ResendUserOtp import ResendUserOtp
 from src.api.models.payload.requests.CreateUserRequest import CreateUserRequest
@@ -233,6 +235,29 @@ async def login(self, req: AuthenticateUserRequest) -> UserSuccess:
             "token": jwt_details,
         }
 
+    async def validate_token(self, req: JWT) -> JWTSuccess:
+        data = self.utility_service.decrypt_jwt(req.token)
+        if not data:
+            message = MESSAGES["AUTH"]["TOKEN_ERROR"]
+            self.logger.info(
+                {
+                    "activity_type": ACTIVITY_TYPES["VALIDATE_TOKEN"],
+                    "message": message,
+                    "metadata": {"token": req.token},
+                }
+            )
+            return {"is_success": False, "message": message}
+
+        message = MESSAGES["AUTH"]["TOKEN_SUCCESS"]
+        self.logger.info(
+            {
+                "activity_type": ACTIVITY_TYPES["VALIDATE_TOKEN"],
+                "message": message,
+                "metadata": {"user": {"email": data["email"], "id": data["user_id"]}},
+            }
+        )
+        return {"is_success": True, "message": message, "data": data}
+
     async def change_password(
         self, id: str, req: ChangeUserPasswordRequest
     ) -> UserSuccess:

src/api/services/UtilityService.py

@@ -8,6 +8,7 @@
 
 from src.env import jwt_config
 from src.utils.svcs import Service
+from src.api.typing.JWT import JWTData
 from src.api.models.postgres import User
 from src.api.typing.ExpireUUID import ExpireUUID
 from src.api.enums.CharacterCasing import CharacterCasing
@@ -70,10 +71,24 @@ def generate_jwt(email: str, user_uuid: str) -> str:
         jwt_claims = {
             "exp": timezone.now() + timedelta(seconds=3600),
             "iss": jwt_config["issuer"],
-            "aud": email,
+            "aud": jwt_config["issuer"],
         }
         return jwt.encode(dict(jwt_data, **jwt_claims), jwt_config["secret"])
 
+    @staticmethod
+    def decrypt_jwt(token: str) -> JWTData | None:
+        try:
+            data = jwt.decode(
+                token,
+                jwt_config["secret"],
+                algorithms=["HS256"],
+                issuer=jwt_config["issuer"],
+                audience=jwt_config["issuer"],
+            )
+            return {"email": data["email"], "user_id": data["user_id"]}
+        except jwt.exceptions.InvalidTokenError:
+            ...
+
     @staticmethod
     def generate_uuid() -> ExpireUUID:
         current_time = timezone.now()

src/api/typing/JWT.py

@@ -0,0 +1,12 @@
+from typing import TypedDict, NotRequired
+
+
+class JWTData(TypedDict):
+    email: str
+    user_id: str
+
+
+class JWTSuccess(TypedDict):
+    is_success: bool
+    message: str
+    data: NotRequired[JWTData]

src/api/utils/error_handlers.py

@@ -1,7 +1,6 @@
 import traceback
 from http import HTTPStatus
 
-import jwt
 from django.http import HttpRequest, HttpResponse
 from ninja.errors import ValidationError, AuthenticationError
 
@@ -15,7 +14,6 @@
 exception_logger = Logger("API Exception")
 
 
-@api.exception_handler(jwt.exceptions.InvalidTokenError)
 @api.exception_handler(AuthenticationError)
 def on_invalid_token(request: HttpRequest, exc: Exception) -> HttpResponse:
     exception_logger.debug(