from jose import jwt import config from auth.atlas.helpers import get_rsa from utils.exceptions import Unauthorized from sentry_sdk import capture_exception from utils.json_encoder import DataclassDictionaryDecoder class AtlasTokenDecoder: def decode(self, token, schema=None, **kwargs): try: payload = jwt.decode( token, get_rsa(), algorithms=config.ATLAS_UM_ALGORITHM, **kwargs, ) if schema: payload = DataclassDictionaryDecoder.convert_from_dict(schema, payload) except jwt.ExpiredSignatureError: raise Unauthorized("Token is expired.") except jwt.JWTClaimsError: raise Unauthorized("Incorrect claims, please check the audience and issuer.") except Exception as ex: capture_exception(ex) raise Unauthorized("Unable to parse authentication token.") return payload