def initialize_connection()

in syndicate/core/__init__.py [0:0]
87 lines of code
9 McCabe index (conditional complexity)

def initialize_connection():
    global CONFIG
    global CONN
    global CONF_PATH
    global CREDENTIALS
    global RESOURCES_PROVIDER
    global PROCESSOR_FACADE

    regex_digest = RegexViewDigest()
    regex_digest.expression = NAMED_S3_URI_PATTERN
    regex_digest.groups = S3_PATTERN_GROUP_NAMES
    uri_bucket_view = URIBucketView()
    uri_bucket_view.digest = regex_digest

    CONFIG = ConfigHolder(CONF_PATH)
    CONFIG.deploy_target_bucket_view = uri_bucket_view
    sts = STSConnection(CONFIG.region, CONFIG.aws_access_key_id,
                        CONFIG.aws_secret_access_key, CONFIG.aws_session_token)
    try:
        CREDENTIALS = {
            'region': CONFIG.region
        }
        if _ready_to_use_provided_temp_creds():
            _LOG.debug(f'Going to use previously generated temporary '
                       f'credentials')
            CREDENTIALS[ACCESS_KEY] = CONFIG.temp_aws_access_key_id
            CREDENTIALS[SECRET_KEY] = CONFIG.temp_aws_secret_access_key
            CREDENTIALS[SESSION_TOKEN] = CONFIG.temp_aws_session_token
        elif _ready_to_assume():
            _LOG.debug('Starting to assume role ...')
            # get CREDENTIALS for N hours

            token_code = prompt_mfa_code() if CONFIG.serial_number else None
            temp_credentials = sts.get_temp_credentials(
                role_arn=CONFIG.access_role,
                acc_id=CONFIG.account_id,
                duration=CONFIG.session_duration,
                serial_number=CONFIG.serial_number,
                token_code=token_code
            )
            _LOG.debug(f'Role {CONFIG.access_role} is assumed successfully'
                       f'for {CONFIG.session_duration} seconds')
            CREDENTIALS[ACCESS_KEY] = temp_credentials['AccessKeyId']
            CREDENTIALS[SECRET_KEY] = temp_credentials['SecretAccessKey']
            CREDENTIALS[SESSION_TOKEN] = temp_credentials['SessionToken']
            _LOG.debug(f'Temporary credentials have been successfully '
                       f'generated by assuming the role {CONFIG.access_role}, '
                       f'saving to config.')
            CONFIG.set_temp_credentials_to_config(
                temp_aws_access_key_id=temp_credentials['AccessKeyId'],
                temp_aws_secret_access_key=temp_credentials['SecretAccessKey'],
                temp_aws_session_token=temp_credentials['SessionToken'],
                expiration=temp_credentials['Expiration']
            )
        elif _ready_to_generate_temp_creds():
            _LOG.debug(f'Going to generate new temporary credentials')

            token_code = None
            if CONFIG.serial_number:
                token_code = prompt_mfa_code()
            temp_credentials = sts.get_session_token(
                duration=CONFIG.session_duration,
                serial_number=CONFIG.serial_number,
                token_code=token_code
            )
            CREDENTIALS[ACCESS_KEY] = temp_credentials['AccessKeyId']
            CREDENTIALS[SECRET_KEY] = temp_credentials['SecretAccessKey']
            CREDENTIALS[SESSION_TOKEN] = temp_credentials['SessionToken']
            _LOG.debug(f'Temporary credentials have been successfully '
                       f'generated, saving to config.')
            CONFIG.set_temp_credentials_to_config(
                temp_aws_access_key_id=temp_credentials['AccessKeyId'],
                temp_aws_secret_access_key=temp_credentials['SecretAccessKey'],
                temp_aws_session_token=temp_credentials['SessionToken'],
                expiration=temp_credentials['Expiration']
            )
        elif _ready_to_use_creds():
            _LOG.debug('Credentials access')
            CREDENTIALS[ACCESS_KEY] = CONFIG.aws_access_key_id
            CREDENTIALS[SECRET_KEY] = CONFIG.aws_secret_access_key
        CONN = ConnectionProvider(CREDENTIALS)
        RESOURCES_PROVIDER = ResourceProvider(config=CONFIG,
                                              credentials=CREDENTIALS,
                                              sts_conn=sts)
        PROCESSOR_FACADE = ProcessorFacade(
            resources_provider=RESOURCES_PROVIDER)
        _LOG.debug('aws-syndicate has been initialized')
    except ClientError as e:
        message = f'An unexpected error has occurred trying to ' \
                  f'init connection: {e}'
        _LOG.error(message)
        raise AssertionError(message)