def _create_credentials()

in alibabacloud_credentials/providers.py [0:0]

• 32 lines of code
• 4 McCabe index (conditional complexity)

    def _create_credentials(self):
        # 获取credential 先实现签名用工具类
        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,
            'AccessKeyId': self.access_key_id,
            'RoleSessionName': self.role_session_name,
            'SignatureMethod': 'HMAC-SHA1',
            'SignatureVersion': '1.0'
        }
        tea_request.query["Timestamp"] = ph.get_iso_8061_date()
        tea_request.query["SignatureNonce"] = ph.get_uuid()
        if self.policy is not None:
            tea_request.query["Policy"] = self.policy
        string_to_sign = ph.compose_string_to_sign("GET", tea_request.query)
        signature = ph.sign_string(string_to_sign, self.access_key_secret + "&")
        tea_request.query["Signature"] = signature
        tea_request.protocol = 'https'
        tea_request.headers['host'] = self.sts_endpoint
        # request
        response = TeaCore.do_action(tea_request)
        if response.status_code == 200:
            dic = json.loads(response.body.decode('utf-8'))
            if "Credentials" in dic:
                cre = dic.get("Credentials")
                # 先转换为时间数组
                time_array = time.strptime(cre.get("Expiration"), "%Y-%m-%dT%H:%M:%SZ")
                # 转换为时间戳
                expiration = calendar.timegm(time_array)
                return credentials.RamRoleArnCredential(cre.get("AccessKeyId"), cre.get("AccessKeySecret"),
                                                        cre.get("SecurityToken"), expiration, self)
        raise CredentialException(response.body.decode('utf-8'))