def token()

in source/openid-waitingroom/chalice/app.py [0:0]

• 42 lines of code
• 8 McCabe index (conditional complexity)

def token():
    """
    This is the token endpoint
    """
    app.log.info('/token')
    app.log.info(app.current_request.to_dict())
    app.log.info(app.current_request.raw_body)
    query_parameters = parse_qs(app.current_request.raw_body.decode("utf-8"))
    try:
        event_id = deep_clean(query_parameters.get("client_id")[0])
        provided_secret = deep_clean(query_parameters.get("client_secret")[0])
        request_id = deep_clean(query_parameters.get("code")[0])
        grant_type = deep_clean(query_parameters.get("grant_type")[0])
        # get the client secret from secrets manager
        valid_client_secret = SECRETS_CLIENT.get_secret_value(
            SecretId=CLIENT_SECRET_ID).get("SecretString")
        if (provided_secret == valid_client_secret) and (
                grant_type in GRANT_TYPES) and (WAITING_ROOM_EVENT_ID
                                                == event_id):
            app.log.info('valid /token request')
            generate_token_api = f'{PRIVATE_API_ENDPOINT}/generate_token'
            # context comes from API Gateway and not the supplied headers
            issuer = (f'https://{app.current_request.context["domainName"]}' +
                      f'/{app.current_request.context["stage"]}')
            body = {
                "event_id": event_id,
                "request_id": request_id,
                "issuer": issuer
            }
            parsed = urlparse(PRIVATE_API_ENDPOINT)
            # create an authentication signer for AWS
            auth = BotoAWSRequestsAuth(aws_host=parsed.netloc,
                                       aws_region=API_REGION,
                                       aws_service='execute-api')
            response = requests.post(generate_token_api, json=body, auth=auth)
            # app.log.info(response.text)
            if response.status_code == 200:
                return Response(status_code=200,
                                body=response.text,
                                headers={'Content-Type': 'application/json'})
            message = f'{response.status_code} status from private API'
            app.log.info(message)
        else:
            app.log.info('invalid /token request')
    except (KeyError, IndexError, TypeError):
        print_exception()
    return Response(status_code=400,
                    body='Bad Request',
                    headers={'Content-Type': 'text/plain'})