auth.py

import json
from flask import request, _request_ctx_stack, abort
from functools import wraps
from jose import jwt
from urllib.request import urlopen

AUTH0_DOMAIN = 'fouz.us.auth0.com'
ALGORITHMS = ['RS256']
API_AUDIENCE = 'capstoneApi'

# AuthError Exception
'''
AuthError Exception
A standardized way to communicate auth failure modes
'''

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


# Auth Header
def get_token_auth_header():
    # get the header from the request
    auth = request.headers.get('Authorization', None)

    # if no header is present
    if not auth:
        raise AuthError({
            'code': 'authorization_header_missing',
            'description': 'Authroization header is expected'}, 401)

    # split the bearer and token
    parts = auth.split()
    # check if the header is malformed
    if parts[0].lower() != 'bearer':
        raise AuthError({
            'code': 'invalid_header',
            'description': 'Authorization header must start with "Bearer"'}, 401)

    elif len(parts) == 1:
        raise AuthError({
            'code': 'invalid_header',
            'description': 'Token not found.'}, 401)

    elif len(parts) > 2:
        raise AuthError({
            'code': 'invaild_header', 'description': 'Authorization header must be bearer token.'}, 401)

    token = parts[1]

    return token


def check_permissions(permission, payload):
    # check if the permissions are not included in the payload
    if 'permissions' not in payload:
        abort(400)
    if permission not in payload['permissions']:
        abort(403)
    # else
    return True


def verify_decode_jwt(token):
    # token: a json web token (string)
    # verify the token using Auth0 /.well
    jsonurl = urlopen(f'https://{AUTH0_DOMAIN}/.well-known/jwks.json')
    jwks = json.loads(jsonurl.read())

    # get user verified_header
    unverified_header = jwt.get_unverified_header(token)

# get the key(secret)
    rsa_key = {}
    if 'kid' not in unverified_header:
        raise AuthError({
            'code': 'invaild header',
            'description': 'Authorization malformed.'}, 401)

    for key in jwks['keys']:
        if key['kid'] == unverified_header['kid']:
            rsa_key = {
                'kty': key['kty'],
                'kid': key['kid'],
                'use': key['use'],
                'n': key['n'],
                'e': key['e']
            }

    # if rsa_key has value decode the playlaod
    if rsa_key:
        try:
            playload = jwt.decode(
                token,
                rsa_key,
                algorithms=ALGORITHMS,
                audience=API_AUDIENCE,
                issuer='https://'+AUTH0_DOMAIN+'/')

            return playload

        except jwt.ExpiredSignatureError:
            raise AuthError({
                'code': 'token_expired',
                'description': 'Token expired.'}, 401)

        except jwt.JWTClaimsError:
            raise AuthError({
                'code': 'invalid_claims',
                'descriptipn': 'Incorrect claims, Please, check the audience and issuer'}, 401)

        except Exception:
            raise AuthError({
                'code': 'invalid_header',
                'description': 'Unable to parse authentication token'}, 400)
        raise AuthError({
            'code': 'invalid_header',
            'description': 'Unable to find the appropriate key'})


def requires_auth(permission=''):
    def requires_auth_decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            token = get_token_auth_header()
            payload = verify_decode_jwt(token)
            check_permissions(permission, payload)
            return f(payload, *args, **kwargs)

        return wrapper
    return requires_auth_decorator