qs.py

import jwt
from jwt import PyJWKClient


class AuthError(Exception):
    def __init__(self, error, status_code):
        self.error = error
        self.status_code = status_code

    @classmethod
    def from_exception(cls, exc):
        if isinstance(exc, jwt.PyJWKClientError):
            return cls(
                {
                    "code": "invalid_header",
                    "description": "Unable to find appropriate key",
                },
                401,
            )

        if isinstance(exc, jwt.ExpiredSignatureError):
            return cls(
                {"code": "token_expired", "description": "token is expired"}, 401
            )

        if isinstance(exc, jwt.JWTClaimsError):
            return cls(
                {
                    "code": "invalid_claims",
                    "description": "incorrect claims, please check the audience and issuer",
                },
                401,
            )

        return cls(
            {
                "code": "invalid_header",
                "description": "Unable to parse authentication token.",
            },
            401,
        )


def requires_auth(f):
    """Determines if the Access Token is valid"""
    url = "https://" + AUTH0_DOMAIN + "/.well-known/jwks.json"
    jwks_client = PyJWKClient(url)

    @wraps(f)
    def decorated(*args, **kwargs):
        token = get_token_auth_header()

        try:
            signing_key = jwks_client.get_signing_key_from_jwt(token)

            payload = jwt.decode(
                token,
                signing_key,
                algorithms=ALGORITHMS,
                audience=API_AUDIENCE,
                issuer="https://" + AUTH0_DOMAIN + "/",
            )
        except Exception as exc:
            raise AuthError.from_exception(exc)

        _request_ctx_stack.top.current_user = payload
        return f(*args, **kwargs)

    return decorated