def retrieve_credentials()

in tts/src/tts/amazonpolly.py [0:0]

• 26 lines of code
• 5 McCabe index (conditional complexity)

    def retrieve_credentials(self):
        try:
            cert_file = self.get_param('certfile')
            key_file = self.get_param('keyfile')
            endpoint = self.get_param('endpoint')
            role_alias = self.get_param('role')
            connect_timeout = self.get_param('connect_timeout_ms', self.DEFAULT_AUTH_CONNECT_TIMEOUT_MS)
            total_timeout = self.get_param('total_timeout_ms', self.DEFAULT_AUTH_TOTAL_TIMEOUT_MS)
            thing_name = self.get_param('thing_name', '')

            if any(v is None for v in (cert_file, key_file, endpoint, role_alias, thing_name)):
                return None

            headers = {'x-amzn-iot-thingname': thing_name} if len(thing_name) > 0 else None
            url = 'https://{}/role-aliases/{}/credentials'.format(endpoint, role_alias)
            timeout = (connect_timeout, total_timeout - connect_timeout)  # see also: urllib3/util/timeout.py

            response = requests.get(url, cert=(cert_file, key_file), headers=headers, timeout=timeout)
            d = response.json()['credentials']

            rospy.loginfo('Credentials expiry time: {}'.format(d['expiration']))

            return {
                'access_key': d['accessKeyId'],
                'secret_key': d['secretAccessKey'],
                'token': d['sessionToken'],
                'expiry_time': d['expiration'],
            }
        except Exception as e:
            rospy.logwarn('Failed to fetch credentials from AWS IoT: {}'.format(e))
            return None