def _refresh_credentials()

in alibabacloud_credentials/provider/ram_role_arn.py [0:0]
53 lines of code
13 McCabe index (conditional complexity)

    def _refresh_credentials(self) -> RefreshResult[Credentials]:
        tea_request = ph.get_new_request()
        tea_request.query = {
            'Action': 'AssumeRole',
            'Format': 'JSON',
            'Version': '2015-04-01',
            'DurationSeconds': str(self._duration_seconds),
            'RoleArn': self._role_arn,
            'RoleSessionName': self._role_session_name,
            'SignatureMethod': 'HMAC-SHA1',
            'SignatureVersion': '1.0',
            'Timestamp': ph.get_iso_8061_date(),
            'SignatureNonce': ph.get_uuid()
        }

        if self._policy is not None and self._policy != '':
            tea_request.query['Policy'] = self._policy

        if self._external_id is not None and self._external_id != '':
            tea_request.query['ExternalId'] = self._external_id

        pre_credentials = self._credentials_provider.get_credentials()
        if pre_credentials is None:
            raise CredentialException('unable to load original credentials from the provider in RAM role arn')

        tea_request.query['AccessKeyId'] = pre_credentials.get_access_key_id()
        security_token = pre_credentials.get_security_token()
        if security_token is not None and security_token != '':
            tea_request.query['SecurityToken'] = security_token

        string_to_sign = ph.compose_string_to_sign('GET', tea_request.query)
        signature = ph.sign_string(string_to_sign, pre_credentials.get_access_key_secret() + '&')
        tea_request.query['Signature'] = signature
        tea_request.protocol = 'https'
        tea_request.headers['host'] = self._sts_endpoint

        response = TeaCore.do_action(tea_request, self._runtime_options)

        if response.status_code != 200:
            raise CredentialException(
                f'error refreshing credentials from ram_role_arn, http_code: {response.status_code}, result: {response.body.decode("utf-8")}')

        dic = json.loads(response.body.decode('utf-8'))
        if 'Credentials' not in dic:
            raise CredentialException(
                f'error retrieving credentials from ram_role_arn result: {response.body.decode("utf-8")}')

        cre = dic.get('Credentials')
        if 'AccessKeyId' not in cre or 'AccessKeySecret' not in cre or 'SecurityToken' not in cre:
            raise CredentialException(
                f'error retrieving credentials from ram_role_arn result: {response.body.decode("utf-8")}')

        # 先转换为时间数组
        time_array = time.strptime(cre.get('Expiration'), '%Y-%m-%dT%H:%M:%SZ')
        # 转换为时间戳
        expiration = calendar.timegm(time_array)
        credentials = Credentials(
            access_key_id=cre.get('AccessKeyId'),
            access_key_secret=cre.get('AccessKeySecret'),
            security_token=cre.get('SecurityToken'),
            expiration=expiration,
            provider_name=f'{self.get_provider_name()}/{pre_credentials.get_provider_name()}'
        )
        return RefreshResult(value=credentials,
                             stale_time=_get_stale_time(expiration))