auth.py

# https://auth0.com/docs/quickstart/backend/python/#validate-access-tokens

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

AUTH0_DOMAIN = 'freefrom-map-api.us.auth0.com'
API_AUDIENCE = 'https://freefrom-map-api.herokuapp.com/'
ALGORITHMS = ['RS256']


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


def get_token_auth_header():
    '''
    Obtains the Access Token from the Authorization Header
    '''
    auth = request.headers.get('Authorization', None)
    if not auth:
        raise AuthError(
            {
                'code': 'authorization_header_missing',
                'description': 'Authorization header is expected',
            }
        )

    parts = auth.split()

    if parts[0].lower() != 'bearer':
        raise AuthError(
            {
                'code': 'invalid_header',
                'description': 'Authorization header must start with Bearer',
            }
        )
    elif len(parts) == 1:
        raise AuthError(
            {
                'code': 'invalid_header',
                'description': 'Token not found',
            }
        )
    elif len(parts) > 2:
        raise AuthError(
            {
                'code': 'invalid_header',
                'description': 'Authorization header must be Bearer token',
            }
        )

    token = parts[1]
    return token


def is_token_valid():
    '''
    Determines if the Access Token is valid
    '''
    token = get_token_auth_header()
    jsonurl = urlopen('https://' + AUTH0_DOMAIN + '/.well-known/jwks.json')
    jwks = json.loads(jsonurl.read())
    unverified_header = jwt.get_unverified_header(token)
    rsa_key = {}
    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:
        try:
            payload = jwt.decode(
                token,
                rsa_key,
                algorithms=ALGORITHMS,
                audience=API_AUDIENCE,
                issuer='https://' + AUTH0_DOMAIN + '/'
            )
        except jwt.ExpiredSignatureError:
            raise AuthError(
                {
                    'code': 'token_expired',
                    'description': 'token is expired',
                }
            )
        except jwt.JWTClaimsError:
            raise AuthError(
                {
                    'code': 'invalid_claims',
                    'description': 'incorrect claims, please check the audience and issuer',
                }
            )
        except Exception:
            raise AuthError(
                {
                    'code': 'invalid_header',
                    'description': 'Unable to parse authentication token.',
                }
            )

        _request_ctx_stack.top.current_user = payload
        return True

    return False


def requires_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        if is_token_valid():
            return f(*args, **kwargs)
        raise AuthError(
            {
                'code': 'invalid_header',
                'description': 'Unable to find appropriate key',
            }
        )
    return decorated