def _get_session_token()

in aws_advanced_python_wrapper/okta_plugin.py [0:0]

• 24 lines of code
• 3 McCabe index (conditional complexity)

    def _get_session_token(self, props: Properties) -> str:
        idp_endpoint = WrapperProperties.IDP_ENDPOINT.get(props)
        idp_user = WrapperProperties.IDP_USERNAME.get(props)
        idp_password = WrapperProperties.IDP_PASSWORD.get(props)

        session_token_endpoint = f"https://{idp_endpoint}/api/v1/authn"

        request_body = {
            "username": idp_user,
            "password": idp_password,
        }
        try:
            r = requests.post(session_token_endpoint,
                              headers={'Content-Type': 'application/json', 'Accept': 'application/json'},
                              json=request_body,
                              verify=WrapperProperties.SSL_SECURE.get_bool(props),
                              timeout=WrapperProperties.HTTP_REQUEST_TIMEOUT.get_int(props))
            if r.status_code / 100 != 2:
                error_message = "SamlUtils.RequestFailed"
                logger.debug(error_message, r.status_code, r.reason, r.text)
                raise AwsWrapperError(Messages.get_formatted(error_message, r.status_code, r.reason, r.text))
            return r.json().get(OktaCredentialsProviderFactory._SESSION_TOKEN)
        except IOError as e:
            error_message = "OktaAuthPlugin.UnhandledException"
            logger.debug(error_message, e)
            raise AwsWrapperError(Messages.get_formatted(error_message, e))