Always set the Date header to the current date

botocore/auth.py

@@ -22,6 +22,7 @@
 from operator import itemgetter
 import functools
 import time
+import calendar
 
 import six
 
@@ -114,8 +115,9 @@ def __init__(self, credentials):
  def add_auth(self, request):
  if self.credentials is None:
  raise NoCredentialsError
- if 'Date' not in request.headers:
- request.headers['Date'] = formatdate(usegmt=True)
+ if 'Date' in request.headers:
+ del request.headers['Date']
+ request.headers['Date'] = formatdate(usegmt=True)
  if self.credentials.token:
  if 'X-Amz-Security-Token' in request.headers:
  del request.headers['X-Amz-Security-Token']
@@ -143,8 +145,8 @@ def __init__(self, credentials, service_name, region_name):
  # We initialize these value here so the unit tests can have
  # valid values. But these will get overriden in ``add_auth``
  # later for real requests.
- now = datetime.datetime.utcnow()
- self.timestamp = now.strftime('%Y%m%dT%H%M%SZ')
+ self._now = datetime.datetime.utcnow()
+ self.timestamp = self._now.strftime('%Y%m%dT%H%M%SZ')
  self._region_name = region_name
  self._service_name = service_name
 
@@ -305,8 +307,8 @@ def add_auth(self, request):
  if self.credentials is None:
  raise NoCredentialsError
  # Create a new timestamp for each signing event
- now = datetime.datetime.utcnow()
- self.timestamp = now.strftime('%Y%m%dT%H%M%SZ')
+ self._now = datetime.datetime.utcnow()
+ self.timestamp = self._now.strftime('%Y%m%dT%H%M%SZ')
  # This could be a retry. Make sure the previous
  # authorization header is removed first.
  self._modify_request_before_signing(request)
@@ -331,15 +333,27 @@ def _inject_signature_to_request(self, request, signature):
  def _modify_request_before_signing(self, request):
  if 'Authorization' in request.headers:
  del request.headers['Authorization']
- if 'Date' not in request.headers:
- if 'X-Amz-Date' in request.headers:
- del request.headers['X-Amz-Date']
- request.headers['X-Amz-Date'] = self.timestamp
+ self._set_necessary_date_headers(request)
  if self.credentials.token:
  if 'X-Amz-Security-Token' in request.headers:
  del request.headers['X-Amz-Security-Token']
  request.headers['X-Amz-Security-Token'] = self.credentials.token
 
+ def _set_necessary_date_headers(self, request):
+ # The spec allows for either the Date _or_ the X-Amz-Date value to be
+ # used so we check both. If there's a Date header, we use the date
+ # header. Otherwise we use the X-Amz-Date header.
+ if 'Date' in request.headers:
+ del request.headers['Date']
+ request.headers['Date'] = formatdate(
+ int(calendar.timegm(self._now.timetuple())))
+ if 'X-Amz-Date' in request.headers:
+ del request.headers['X-Amz-Date']
+ else:
+ if 'X-Amz-Date' in request.headers:
+ del request.headers['X-Amz-Date']
+ request.headers['X-Amz-Date'] = self.timestamp
+
 
 class S3SigV4Auth(SigV4Auth):
 
@@ -472,8 +486,9 @@ def sign_string(self, string_to_sign):
  def canonical_standard_headers(self, headers):
  interesting_headers = ['content-md5', 'content-type', 'date']
  hoi = []
- if 'Date' not in headers:
- headers['Date'] = formatdate(usegmt=True)
+ if 'Date' in headers:
+ del headers['Date']
+ headers['Date'] = formatdate(usegmt=True)
  for ih in interesting_headers:
  found = False
  for key in headers:

tests/unit/auth/test_signers.py

@@ -36,6 +36,12 @@ def setUp(self):
  self.credentials = botocore.credentials.Credentials(access_key,
  secret_key)
  self.hmacv1 = botocore.auth.HmacV1Auth(self.credentials, None, None)
+ self.date_mock = mock.patch('botocore.auth.formatdate')
+ self.formatdate = self.date_mock.start()
+ self.formatdate.return_value = 'Thu, 17 Nov 2005 18:49:58 GMT'
+
+ def tearDown(self):
+ self.date_mock.stop()
 
  def test_put(self):
  headers = {'Date': 'Thu, 17 Nov 2005 18:49:58 GMT',
@@ -115,11 +121,37 @@ def test_resign_with_token(self):
  auth.add_auth(request)
  original_auth = request.headers['Authorization']
  # Resigning the request shouldn't change the authorization
- # header.
+ # header. We are also ensuring that the date stays the same
+ # because we're mocking out the formatdate() call. There's
+ # another unit test that verifies we use the latest time
+ # when we sign the request.
  auth.add_auth(request)
  self.assertEqual(request.headers.get_all('Authorization'),
  [original_auth])
 
+ def test_resign_uses_most_recent_date(self):
+ dates = [
+ 'Thu, 17 Nov 2005 18:49:58 GMT',
+ 'Thu, 17 Nov 2014 20:00:00 GMT',
+ ]
+ self.formatdate.side_effect = dates
+
+ request = AWSRequest()
+ request.headers['Content-Type'] = 'text/html'
+ request.method = 'PUT'
+ request.url = 'https://s3.amazonaws.com/bucket/key'
+
+ self.hmacv1.add_auth(request)
+ original_date = request.headers['Date']
+
+ self.hmacv1.add_auth(request)
+ modified_date = request.headers['Date']
+
+ # Each time we sign a request, we make another call to formatdate()
+ # so we should have a different date header each time.
+ self.assertEqual(original_date, dates[0])
+ self.assertEqual(modified_date, dates[1])
+
 
 class TestSigV2(unittest.TestCase):
 
@@ -177,6 +209,12 @@ def setUp(self):
  self.credentials = botocore.credentials.Credentials(self.access_key,
  self.secret_key)
  self.auth = botocore.auth.SigV3Auth(self.credentials)
+ self.date_mock = mock.patch('botocore.auth.formatdate')
+ self.formatdate = self.date_mock.start()
+ self.formatdate.return_value = 'Thu, 17 Nov 2005 18:49:58 GMT'
+
+ def tearDown(self):
+ self.date_mock.stop()
 
  def test_signature_with_date_headers(self):
  request = AWSRequest()

tests/unit/auth/test_sigv4.py

@@ -100,13 +100,19 @@ def test_generator():
  )
  mocked_datetime = datetime_patcher.start()
  mocked_datetime.utcnow.return_value = datetime.datetime(2011, 9, 9, 23, 36)
+ formatdate_patcher = mock.patch('botocore.auth.formatdate')
+ formatdate = formatdate_patcher.start()
+ # We have to change this because Sep 9, 2011 was actually
+ # a Friday, but the tests have this set to a Monday.
+ formatdate.return_value = 'Mon, 09 Sep 2011 23:36:00 GMT'
  for test_case in set(os.path.splitext(i)[0]
  for i in os.listdir(TESTSUITE_DIR)):
  if test_case in TESTS_TO_IGNORE:
  log.debug("Skipping test: %s", test_case)
  continue
  yield (_test_signature_version_4, test_case)
  datetime_patcher.stop()
+ formatdate_patcher.stop()
 
 
 def create_request_from_raw_request(raw_request):