<commit_before>"""
.. module: lemur.auth.views
    :platform: Unix
    :copyright: (c) 2018 by Netflix Inc., see AUTHORS for more
    :license: Apache, see LICENSE for more details.
.. moduleauthor:: <NAME> <<EMAIL>>
"""
import jwt
import base64
import requests

from flask import Blueprint, current_app

from flask_restful import reqparse, Resource, Api
from flask_principal import Identity, identity_changed

from lemur.constants import SUCCESS_METRIC_STATUS, FAILURE_METRIC_STATUS
from lemur.extensions import metrics
from lemur.common.utils import get_psuedo_random_string

from lemur.users import service as user_service
from lemur.roles import service as role_service
from lemur.auth.service import create_token, fetch_token_header, get_rsa_public_key
from lemur.auth import ldap


mod = Blueprint('auth', __name__)
api = Api(mod)


def exchange_for_access_token(code, redirect_uri, client_id, secret, access_token_url=None, verify_cert=True):
    """
    Exchanges authorization code for access token.

    :param code:
    :param redirect_uri:
    :param client_id:
    :param secret:
    :param access_token_url:
    :param verify_cert:
    :return:
    :return:
    """
    # take the information we have received from the provider to create a new request
    params = {
        'grant_type': 'authorization_code',
        'scope': 'openid email profile address',
        'code': code,
        'redirect_uri': redirect_uri,
        'client_id': client_id
    }

    # the secret and cliendId will be given to you when you signup for the provider
    token = '{0}:{1}'.format(client_id, secret)

    basic = base64.b64encode(bytes(token, 'utf-8'))
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'authorization': 'basic {0}'.format(basic.decode('utf-8'))
    }

    # exchange authorization code for access token.
    r = requests.post(access_token_url, headers=headers, params=params, verify=verify_cert)
    if r.status_code == 400:
        r = requests.post(access_token_url, headers=headers, data=params, verify=verify_cert)
    id_token = r.json()['id_token']
    access_token = r.json()['access_token']

    return id_token, access_token


def validate_id_token(id_token, client_id, jwks_url):
    """
    Ensures that the token we receive is valid.

    :param id_token:
    :param client_id:
    :param jwks_url:
    :return:
    """
    # fetch token public key
    header_data = fetch_token_header(id_token)

    # retrieve the key material as specified by the token header
    r = requests.get(jwks_url)
    for key in r.json()['keys']:
        if key['kid'] == header_data['kid']:
            secret = get_rsa_public_key(key['n'], key['e'])
            algo = header_data['alg']
            break
    else:
        return dict(message='Key not found'), 401

    # validate your token based on the key it was signed with
    try:
        jwt.decode(id_token, secret.decode('utf-8'), algorithms=[algo], audience=client_id)
    except jwt.DecodeError:
        return dict(message='Token is invalid'), 401
    except jwt.ExpiredSignatureError:
        return dict(message='Token has expired'), 401
    except jwt.InvalidTokenError:
        return dict(message='Token is invalid'), 401


def retrieve_user(user_api_url, access_token):
    """
    Fetch user information from provided user api_url.

    :param user_api_url:
    :param access_token:
    :return:
    """
    user_params = dict(access_token=access_token, schema='profile')

    # retrieve information about the current user.
    r = requests.get(
        user_api_url,
        params=user_params,
        headers={'Authorization': 'Bearer {}'.format(access_token)})<commit_msg>Converting userinfo authorization to a config var<commit_after>"""
.. module: lemur.auth.views
    :platform: Unix
    :copyright: (c) 2018 by Netflix Inc., see AUTHORS for more
    :license: Apache, see LICENSE for more details.
.. moduleauthor:: <NAME> <<EMAIL>>
"""
import jwt
import base64
import requests

from flask import Blueprint, current_app

from flask_restful import reqparse, Resource, Api
from flask_principal import Identity, identity_changed

from lemur.constants import SUCCESS_METRIC_STATUS, FAILURE_METRIC_STATUS
from lemur.extensions import metrics
from lemur.common.utils import get_psuedo_random_string

from lemur.users import service as user_service
from lemur.roles import service as role_service
from lemur.auth.service import create_token, fetch_token_header, get_rsa_public_key
from lemur.auth import ldap


mod = Blueprint('auth', __name__)
api = Api(mod)


def exchange_for_access_token(code, redirect_uri, client_id, secret, access_token_url=None, verify_cert=True):
    """
    Exchanges authorization code for access token.

    :param code:
    :param redirect_uri:
    :param client_id:
    :param secret:
    :param access_token_url:
    :param verify_cert:
    :return:
    :return:
    """
    # take the information we have received from the provider to create a new request
    params = {
        'grant_type': 'authorization_code',
        'scope': 'openid email profile address',
        'code': code,
        'redirect_uri': redirect_uri,
        'client_id': client_id
    }

    # the secret and cliendId will be given to you when you signup for the provider
    token = '{0}:{1}'.format(client_id, secret)

    basic = base64.b64encode(bytes(token, 'utf-8'))
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'authorization': 'basic {0}'.format(basic.decode('utf-8'))
    }

    # exchange authorization code for access token.
    r = requests.post(access_token_url, headers=headers, params=params, verify=verify_cert)
    if r.status_code == 400:
        r = requests.post(access_token_url, headers=headers, data=params, verify=verify_cert)
    id_token = r.json()['id_token']
    access_token = r.json()['access_token']

    return id_token, access_token


def validate_id_token(id_token, client_id, jwks_url):
    """
    Ensures that the token we receive is valid.

    :param id_token:
    :param client_id:
    :param jwks_url:
    :return:
    """
    # fetch token public key
    header_data = fetch_token_header(id_token)

    # retrieve the key material as specified by the token header
    r = requests.get(jwks_url)
    for key in r.json()['keys']:
        if key['kid'] == header_data['kid']:
            secret = get_rsa_public_key(key['n'], key['e'])
            algo = header_data['alg']
            break
    else:
        return dict(message='Key not found'), 401

    # validate your token based on the key it was signed with
    try:
        jwt.decode(id_token, secret.decode('utf-8'), algorithms=[algo], audience=client_id)
    except jwt.DecodeError:
        return dict(message='Token is invalid'), 401
    except jwt.ExpiredSignatureError:
        return dict(message='Token has expired'), 401
    except jwt.InvalidTokenError:
        return dict(message='Token is invalid'), 401


def retrieve_user(user_api_url, access_token):
    """
    Fetch user information from provided user api_url.

    :param user_api_url:
    :param access_token:
    :return:
    """
    user_params = dict(access_token=access_token, schema='profile')

    headers = {}

    if current_app.config.get('PING_INCLUDE_BEARER_TOKEN'):
        headers = {'Authorization': f'Bearer {access_token}'}

    # retrieve information about the current user.
    r = requests.get(
        user_api_url,
        params=user_params,
        headers=headers,
    )
{'fraction_non_alphanumeric': 0.10160711913932793, 'fraction_numerical': 0.008234825342010891, 'mean_word_length': 3.4955223880597015, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2442079', 'n_tokens_mistral': 2382, 'n_tokens_neox': 2273, 'n_words': 648}