I’m working on an application with FastApi and Python. And I’m consuming the jwt of another application that was developed in Django, which is only used for project authentication in general.
In some other projects, I have already done this authentication, but in this new one, it is not working.
This is my auth_utils.py
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Request
from fastapi.exceptions import HTTPException
import jwt
import base64
from jwt.exceptions import InvalidTokenError
import json
from loguru import logger
import os
ALGORITHM = os.getenv('ALGORITHM')
JWT_SECRET_KEY = os.getenv('SECRET')
def decodeJWT(jwtoken: str):
jwt_options = {
'verify_signature': True,
'verify_exp': True,
'verify_nbf': False,
'verify_iat': True,
'verify_aud': False
}
try:
payload = json.loads(json.dumps(jwt.decode(jwt=jwtoken,
key=JWT_SECRET_KEY,
algorithms=ALGORITHM,
options=jwt_options)))
print(payload)
return payload
except InvalidTokenError as e:
logger.error(f'{e}')
return None
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 self.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.")
def verify_jwt(self, jwtoken: str) -> bool:
isTokenValid: bool = False
try:
payload = decodeJWT(jwtoken)
except:
payload = None
if payload:
isTokenValid = True
return isTokenValid
jwt_bearer = JWTBearer()
def format_jwt(token):
token_decode = token.split(".")[1]
padding = '=' * (4 - len(token_decode) % 4)
token_decode += padding
decoded_bytes = base64.urlsafe_b64decode(token_decode)
decoded_str = decoded_bytes.decode('utf-8')
decoded_payload = json.loads(decoded_str)
return decoded_payload
This here is an example of an endpoint, which I am consuming the authentication function
from api.v1.apps.theme.service.services import insert, get_all, get_one, update, remove
from fastapi import APIRouter, HTTPException, status, Depends
from sqlalchemy.exc import SQLAlchemyError, IntegrityError
from api.v1.apps.theme.schemas.schemas import ThemeSchema
from helpers.auth_utils import JWTBearer, format_jwt
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.exc import NoResultFound
from db.session import get_async_session
from loguru import logger
import os
router = APIRouter()
SUPER_ADMIN = os.getenv('SUPER_ADMIN')
ADMIN = os.getenv('ADMIN')
VIEWER = os.getenv('VIEWER')
@router.post('/create-theme/')
async def create_theme(theme: ThemeSchema, dependencies=Depends(JWTBearer()), session: AsyncSession = Depends(get_async_session)):
"""Endpoint para criação de temas"""
try:
decoded_payload = format_jwt(dependencies)
usuario = decoded_payload.get('role')
if usuario == SUPER_ADMIN or usuario == ADMIN:
return await insert(description=theme.description)
except IntegrityError as e:
logger.error(f"Integrity error when saving theme: {e}")
await session.rollback()
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Integrity violation when saving theme")
except SQLAlchemyError as e:
logger.error(f"Database error when saving theme: {e}")
await session.rollback()
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error when saving theme")
except Exception as e:
logger.error(f"Unexpected error when saving theme: {e}")
await session.rollback()
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Error saving theme")
This is my logger of the error output:
2024-06-21T16:16:36.975431-0300 Signature verification failed