Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if 'x-amz-date' not in headers:
headers['x-amz-date'] = \
datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S +0000')
if self.iam_token is not None:
self._refresh_iam_credentials()
headers['x-amz-security-token'] = self.iam_token
headers['Authorization'] = self.sign(method, bucket_name, url, headers,
sub_resources)
while True:
try:
if self.connection is None:
self.connection = httplib.HTTPConnection(self.host)
self.connection.request(method, url, body, headers)
response = self.connection.getresponse()
if response.status not in xrange(200, 300):
raise S3Error(method, url, body, headers, response)
headers = HeaderDict(response.getheaders())
body = response.read()
return (headers, body)
except httplib.BadStatusLine as exc:
logger.warn(exc)
self.connection = None
except httplib.CannotSendRequest as exc:
logger.warn(exc)
self.connection = None
except ssl.SSLError as exc:
logger.warn(exc)
self.connection = None
except IOError as exc:
if exc.errno == errno.ECONNRESET:
logger.warn(exc)
self.connection = None