x
/
swift3
Code Issues Proposed changes

Fix date validation 

According to [1] when an Authorization header is specified, either a
Date or x-amz-date header needs to be specified, with the x-amz-date
header taking precedence.

Now, the x-amz-date header is validated first, and if both headers are
missing, an AccessDenied error should be returned.  This should prevent
replay attacks occurring on valid requests that are missing the Date
header.

[1]
http://docs.aws.amazon.com/AmazonS3/latest/API/RESTCommonRequestHeaders.
html

N.B. This also fixes some pylint issues and dependencies

Closes-Bug: 1497424
SecurityImpact
[CVE-2015-8466]

Co-Authored-By: Darryl Tam <dtam@swiftstack.com>
Co-Authored-By: Tim Burke <tim.burke@gmail.com>

Change-Id: Ibeff8503fa147e1cf08c1b5374aecee7a4c0bee2
This commit is contained in:
Kota Tsuyuzaki 2015-11-25 14:16:06 -08:00
parent b90a6615d5
commit 4fce274c50
16 changed files with 386 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
swift3/request.py
View File

@ -186,9 +186,11 @@ class Request(swob.Request):
raise InvalidArgument('Content-Length',
self.environ['CONTENT_LENGTH'])
if 'Date' in self.headers:
date_header = self.headers.get('x-amz-date',
self.headers.get('Date', None))
if date_header:
now = datetime.datetime.utcnow()
date = email.utils.parsedate(self.headers['Date'])
date = email.utils.parsedate(date_header)
if 'Expires' in self.params:
try:
d = email.utils.formatdate(float(self.params['Expires']))
@ -213,7 +215,11 @@ class Request(swob.Request):
if abs(d1 - now) > delta:
raise RequestTimeTooSkewed()
else:
raise AccessDenied()
raise AccessDenied('AWS authentication requires a valid Date '
'or x-amz-date header')
else:
raise AccessDenied('AWS authentication requires a valid Date '
'or x-amz-date header')
if 'Content-MD5' in self.headers:
value = self.headers['Content-MD5']

2
swift3/subresource.py
View File

@ -355,7 +355,7 @@ class Grant(object):
if permission.upper() not in PERMISSIONS:
raise S3NotImplemented()
if not isinstance(grantee, Grantee):
raise
raise ValueError()
self.grantee = grantee
self.permission = permission

9
swift3/test/unit/__init__.py
View File

@ -14,6 +14,9 @@
# limitations under the License.
import unittest
from datetime import datetime
import email
import time
from swift.common import swob
@ -90,12 +93,16 @@ class Swift3TestCase(unittest.TestCase):
uri += path
self.swift.register(method, uri, response_class, headers, None)
headers.update({'Authorization': 'AWS test:tester:hmac'})
headers.update({'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
req = swob.Request.blank(path, environ={'REQUEST_METHOD': method},
headers=headers)
status, headers, body = self.call_swift3(req)
return self._get_error_code(body)
def get_date_header(self):
return email.utils.formatdate(time.mktime(datetime.now().timetuple()))
def call_app(self, req, app=None, expect_exception=False):
if app is None:
app = self.app

16
swift3/test/unit/test_acl.py
View File

@ -44,7 +44,8 @@ class TestSwift3Acl(Swift3TestCase):
def test_bucket_acl_GET(self):
req = Request.blank('/bucket?acl',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self._check_acl('test:tester', body)
@ -63,7 +64,8 @@ class TestSwift3Acl(Swift3TestCase):
xml = tostring(elem)
req = Request.blank('/bucket?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body=xml)
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -72,6 +74,7 @@ class TestSwift3Acl(Swift3TestCase):
environ={'REQUEST_METHOD': 'PUT',
'wsgi.input': StringIO(xml)},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'Transfer-Encoding': 'chunked'})
self.assertIsNone(req.content_length)
self.assertIsNone(req.message_length())
@ -82,6 +85,7 @@ class TestSwift3Acl(Swift3TestCase):
req = Request.blank('/bucket?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'X-AMZ-ACL': 'public-read'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -91,6 +95,7 @@ class TestSwift3Acl(Swift3TestCase):
req = Request.blank('/bucket?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'X-AMZ-ACL': 'public-read'})
with mock.patch('swift3.request.handle_acl_header') as mock_handler:
status, headers, body = self.call_swift3(req)
@ -113,6 +118,7 @@ class TestSwift3Acl(Swift3TestCase):
req = Request.blank('/bucket?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'X-AMZ-ACL': 'public-read'},
body=xml)
status, headers, body = self.call_swift3(req)
@ -122,14 +128,16 @@ class TestSwift3Acl(Swift3TestCase):
def test_object_acl_GET(self):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self._check_acl('test:tester', body)
def test_invalid_xml(self):
req = Request.blank('/bucket?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body='invalid')
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'MalformedACLError')

75
swift3/test/unit/test_bucket.py
View File

@ -63,14 +63,16 @@ class TestSwift3Bucket(Swift3TestCase):
def test_bucket_HEAD(self):
req = Request.blank('/junk',
environ={'REQUEST_METHOD': 'HEAD'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
def test_bucket_HEAD_error(self):
req = Request.blank('/nojunk',
environ={'REQUEST_METHOD': 'HEAD'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '404')
self.assertEquals(body, '') # sanifty
@ -78,14 +80,16 @@ class TestSwift3Bucket(Swift3TestCase):
def test_bucket_HEAD_slash(self):
req = Request.blank('/junk/',
environ={'REQUEST_METHOD': 'HEAD'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
def test_bucket_HEAD_slash_error(self):
req = Request.blank('/nojunk/',
environ={'REQUEST_METHOD': 'HEAD'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '404')
@ -104,7 +108,8 @@ class TestSwift3Bucket(Swift3TestCase):
bucket_name = 'junk'
req = Request.blank('/%s' % bucket_name,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -129,7 +134,8 @@ class TestSwift3Bucket(Swift3TestCase):
bucket_name = 'junk-subdir'
req = Request.blank('/%s' % bucket_name,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
elem = fromstring(body, 'ListBucketResult')
@ -147,14 +153,16 @@ class TestSwift3Bucket(Swift3TestCase):
req = Request.blank('/%s?max-keys=5' % bucket_name,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListBucketResult')
self.assertEquals(elem.find('./IsTruncated').text, 'false')
req = Request.blank('/%s?max-keys=4' % bucket_name,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListBucketResult')
self.assertEquals(elem.find('./IsTruncated').text, 'true')
@ -164,7 +172,8 @@ class TestSwift3Bucket(Swift3TestCase):
req = Request.blank('/%s?max-keys=5' % bucket_name,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListBucketResult')
self.assertEquals(elem.find('./MaxKeys').text, '5')
@ -175,7 +184,8 @@ class TestSwift3Bucket(Swift3TestCase):
req = Request.blank('/%s?max-keys=5000' % bucket_name,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListBucketResult')
self.assertEquals(elem.find('./MaxKeys').text, '5000')
@ -189,7 +199,8 @@ class TestSwift3Bucket(Swift3TestCase):
req = Request.blank('/%s?max-keys=invalid' % bucket_name,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -198,7 +209,8 @@ class TestSwift3Bucket(Swift3TestCase):
req = Request.blank('/%s?max-keys=-1' % bucket_name,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -208,7 +220,8 @@ class TestSwift3Bucket(Swift3TestCase):
req = Request.blank('/%s?max-keys=%s' %
(bucket_name, MAX_32BIT_INT + 1),
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -216,7 +229,8 @@ class TestSwift3Bucket(Swift3TestCase):
bucket_name = 'junk'
req = Request.blank('/%s?delimiter=a&marker=b&prefix=c' % bucket_name,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListBucketResult')
self.assertEquals(elem.find('./Prefix').text, 'c')
@ -235,7 +249,8 @@ class TestSwift3Bucket(Swift3TestCase):
'/%s?delimiter=\xef\xbc\xa1&marker=\xef\xbc\xa2&'
'prefix=\xef\xbc\xa3' % bucket_name,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListBucketResult')
self.assertEquals(elem.find('./Prefix').text, '\xef\xbc\xa3')
@ -252,7 +267,8 @@ class TestSwift3Bucket(Swift3TestCase):
bucket_name = 'junk'
req = Request.blank('/%s?delimiter=a&max-keys=2' % bucket_name,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
elem = fromstring(body, 'ListBucketResult')
@ -264,7 +280,8 @@ class TestSwift3Bucket(Swift3TestCase):
bucket_name = 'junk-subdir'
req = Request.blank('/%s?delimiter=a&max-keys=1' % bucket_name,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
elem = fromstring(body, 'ListBucketResult')
@ -313,7 +330,8 @@ class TestSwift3Bucket(Swift3TestCase):
def test_bucket_PUT(self):
req = Request.blank('/bucket',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
self.assertEquals(headers['Location'], '/bucket')
@ -323,6 +341,7 @@ class TestSwift3Bucket(Swift3TestCase):
req = Request.blank('/bucket',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'Transfer-Encoding': 'chunked'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -336,7 +355,8 @@ class TestSwift3Bucket(Swift3TestCase):
req = Request.blank('/bucket',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body=xml)
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -345,6 +365,7 @@ class TestSwift3Bucket(Swift3TestCase):
req = Request.blank('/bucket',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'X-Amz-Acl': 'public-read'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -361,6 +382,7 @@ class TestSwift3Bucket(Swift3TestCase):
req = Request.blank('/bucket',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'X-Amz-Acl': 'public-read'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -378,7 +400,8 @@ class TestSwift3Bucket(Swift3TestCase):
req = Request.blank('/bucket',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body=xml)
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body),
@ -388,7 +411,8 @@ class TestSwift3Bucket(Swift3TestCase):
def test_bucket_PUT_with_location_invalid_xml(self):
req = Request.blank('/bucket',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body='invalid_xml')
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'MalformedXML')
@ -412,14 +436,16 @@ class TestSwift3Bucket(Swift3TestCase):
def test_bucket_DELETE(self):
req = Request.blank('/bucket',
environ={'REQUEST_METHOD': 'DELETE'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '204')
def _test_bucket_for_s3acl(self, method, account):
req = Request.blank('/bucket',
environ={'REQUEST_METHOD': method},
headers={'Authorization': 'AWS %s:hmac' % account})
headers={'Authorization': 'AWS %s:hmac' % account,
'Date': self.get_date_header()})
return self.call_swift3(req)
@ -450,7 +476,8 @@ class TestSwift3Bucket(Swift3TestCase):
def _test_bucket_GET_canned_acl(self, bucket):
req = Request.blank('/%s' % bucket,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
return self.call_swift3(req)

6
swift3/test/unit/test_location.py
View File

@ -35,7 +35,8 @@ class TestSwift3Location(Swift3TestCase):
def test_object_location(self):
req = Request.blank('/bucket?location',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
elem = fromstring(body, 'LocationConstraint')
@ -46,7 +47,8 @@ class TestSwift3Location(Swift3TestCase):
CONF.location = 'us-west-1'
req = Request.blank('/bucket?location',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
elem = fromstring(body, 'LocationConstraint')

12
swift3/test/unit/test_logging.py
View File

@ -29,7 +29,8 @@ class TestSwift3Logging(Swift3TestCase):
def test_bucket_logging_GET(self):
req = Request.blank('/bucket?logging',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
xml = fromstring(body, 'BucketLoggingStatus')
self.assertEquals(xml.keys(), [])
@ -38,14 +39,16 @@ class TestSwift3Logging(Swift3TestCase):
def test_object_logging_GET_error(self):
req = Request.blank('/bucket/object?logging',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'NoLoggingStatusForKey')
def test_bucket_logging_PUT(self):
req = Request.blank('/bucket?logging',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
# FIXME: Support PUT logging
# self.assertEquals(status, 201)
@ -54,7 +57,8 @@ class TestSwift3Logging(Swift3TestCase):
def test_object_logging_PUT_error(self):
req = Request.blank('/bucket/object?logging',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'NoLoggingStatusForKey')

72
swift3/test/unit/test_middleware.py
View File

@ -45,14 +45,16 @@ class TestSwift3Middleware(Swift3TestCase):
def test_bad_format_authorization(self):
req = Request.blank('/something',
headers={'Authorization': 'hoge'})
headers={'Authorization': 'hoge',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'AccessDenied')
def test_bad_method(self):
req = Request.blank('/',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'MethodNotAllowed')
@ -63,7 +65,8 @@ class TestSwift3Middleware(Swift3TestCase):
swob.HTTPOk, {}, None)
req = Request.blank('/%s/%s' % (bucket_name, object_name),
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
raw_path_info = "/%s/%s" % (bucket_name, object_name)
path_info = req.environ['PATH_INFO']
@ -81,12 +84,13 @@ class TestSwift3Middleware(Swift3TestCase):
else:
query_string = ''
req = S3Request({
'REQUEST_METHOD': 'GET',
'PATH_INFO': path,
'QUERY_STRING': query_string,
'HTTP_AUTHORIZATION': 'AWS X:Y:Z',
})
with patch('swift3.request.Request._validate_headers'):
req = S3Request({
'REQUEST_METHOD': 'GET',
'PATH_INFO': path,
'QUERY_STRING': query_string,
'HTTP_AUTHORIZATION': 'AWS X:Y:Z',
})
req.headers.update(headers)
return req._canonical_string()
@ -154,7 +158,8 @@ class TestSwift3Middleware(Swift3TestCase):
expire = '1000000000'
req = Request.blank('/bucket/object?Signature=X&Expires=%s&'
'AWSAccessKeyId=test:tester' % expire,
environ={'REQUEST_METHOD': 'GET'})
environ={'REQUEST_METHOD': 'GET'},
headers={'Date': self.get_date_header()})
req.headers['Date'] = datetime.utcnow()
req.content_type = 'text/plain'
status, headers, body = self.call_swift3(req)
@ -164,7 +169,8 @@ class TestSwift3Middleware(Swift3TestCase):
expire = '10000000000'
req = Request.blank('/bucket/object?Signature=X&Expires=%s&'
'AWSAccessKeyId=test:tester' % expire,
environ={'REQUEST_METHOD': 'GET'})
environ={'REQUEST_METHOD': 'GET'},
headers={'Date': self.get_date_header()})
req.headers['Date'] = datetime.utcnow()
req.content_type = 'text/plain'
status, headers, body = self.call_swift3(req)
@ -177,7 +183,8 @@ class TestSwift3Middleware(Swift3TestCase):
expire = 'invalid'
req = Request.blank('/bucket/object?Signature=X&Expires=%s&'
'AWSAccessKeyId=test:tester' % expire,
environ={'REQUEST_METHOD': 'GET'})
environ={'REQUEST_METHOD': 'GET'},
headers={'Date': self.get_date_header()})
req.headers['Date'] = datetime.utcnow()
req.content_type = 'text/plain'
status, headers, body = self.call_swift3(req)
@ -187,7 +194,8 @@ class TestSwift3Middleware(Swift3TestCase):
expire = 'invalid'
req = Request.blank('/bucket/object?Expires=%s&'
'AWSAccessKeyId=test:tester' % expire,
environ={'REQUEST_METHOD': 'GET'})
environ={'REQUEST_METHOD': 'GET'},
headers={'Date': self.get_date_header()})
req.headers['Date'] = datetime.utcnow()
req.content_type = 'text/plain'
status, headers, body = self.call_swift3(req)
@ -198,7 +206,8 @@ class TestSwift3Middleware(Swift3TestCase):
environ={'HTTP_HOST': 'bucket.localhost:80',
'REQUEST_METHOD': 'HEAD',
'HTTP_AUTHORIZATION':
'AWS test:tester:hmac'})
'AWS test:tester:hmac'},
headers={'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -207,7 +216,8 @@ class TestSwift3Middleware(Swift3TestCase):
environ={'HTTP_HOST': 'bucket.localhost:80',
'REQUEST_METHOD': 'HEAD',
'HTTP_AUTHORIZATION':
'AWS test:tester:hmac'})
'AWS test:tester:hmac'},
headers={'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -222,16 +232,20 @@ class TestSwift3Middleware(Swift3TestCase):
'&partNumber=1',
environ={'REQUEST_METHOD': 'PUT'})
req.headers['Authorization'] = 'AWS test:tester:hmac'
date_header = self.get_date_header()
req.headers['Date'] = date_header
status, headers, body = self.call_swift3(req)
_, _, headers = self.swift.calls_with_headers[-1]
self.assertEquals(base64.urlsafe_b64decode(
headers['X-Auth-Token']),
'PUT\n\n\n/bucket/object?partNumber=1&uploadId=123456789abcdef')
'PUT\n\n\n%s\n/bucket/object?partNumber=1&uploadId=123456789abcdef'
% date_header)
def test_invalid_uri(self):
req = Request.blank('/bucket/invalid\xffname',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidURI')
@ -239,7 +253,8 @@ class TestSwift3Middleware(Swift3TestCase):
req = Request.blank('/bucket/object',
environ={'REQUEST_METHOD': 'PUT',
'HTTP_AUTHORIZATION': 'AWS X:Y:Z',
'HTTP_CONTENT_MD5': '#'})
'HTTP_CONTENT_MD5': '#'},
headers={'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidDigest')
@ -250,7 +265,8 @@ class TestSwift3Middleware(Swift3TestCase):
'/bucket/object',
environ={'REQUEST_METHOD': 'PUT',
'HTTP_AUTHORIZATION': 'AWS X:Y:Z',
'HTTP_CONTENT_MD5': md5_str})
'HTTP_CONTENT_MD5': md5_str},
headers={'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidDigest')
@ -261,7 +277,8 @@ class TestSwift3Middleware(Swift3TestCase):
'/bucket/object',
environ={'REQUEST_METHOD': 'PUT',
'HTTP_AUTHORIZATION': 'AWS X:Y:Z',
'HTTP_CONTENT_MD5': md5_str})
'HTTP_CONTENT_MD5': md5_str},
headers={'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidDigest')
@ -270,7 +287,8 @@ class TestSwift3Middleware(Swift3TestCase):
environ={'REQUEST_METHOD': 'GET',
'HTTP_AUTHORIZATION': 'AWS X:Y:Z',
'HTTP_X_AMZ_METADATA_DIRECTIVE':
'invalid'})
'invalid'},
headers={'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -278,7 +296,8 @@ class TestSwift3Middleware(Swift3TestCase):
req = Request.blank('/',
environ={'REQUEST_METHOD': 'GET',
'HTTP_AUTHORIZATION': 'AWS X:Y:Z',
'HTTP_X_AMZ_STORAGE_CLASS': 'INVALID'})
'HTTP_X_AMZ_STORAGE_CLASS': 'INVALID'},
headers={'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidStorageClass')
@ -286,7 +305,8 @@ class TestSwift3Middleware(Swift3TestCase):
req = Request.blank('/error',
environ={'REQUEST_METHOD': 'GET',
'HTTP_AUTHORIZATION': 'AWS X:Y:Z'},
headers={'x-amz-' + header: 'value'})
headers={'x-amz-' + header: 'value',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'NotImplemented')
@ -302,7 +322,8 @@ class TestSwift3Middleware(Swift3TestCase):
def _test_unsupported_resource(self, resource):
req = Request.blank('/error?' + resource,
environ={'REQUEST_METHOD': 'GET',
'HTTP_AUTHORIZATION': 'AWS X:Y:Z'})
'HTTP_AUTHORIZATION': 'AWS X:Y:Z'},
headers={'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'NotImplemented')
@ -333,7 +354,8 @@ class TestSwift3Middleware(Swift3TestCase):
def test_unsupported_method(self):
req = Request.blank('/bucket?acl',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'Error')
self.assertEquals(elem.find('./Code').text, 'MethodNotAllowed')

10
swift3/test/unit/test_multi_delete.py
View File

@ -42,6 +42,7 @@ class TestSwift3MultiDelete(Swift3TestCase):
req = Request.blank('/bucket/object?delete',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'Content-MD5': content_md5},
body=body)
@ -65,6 +66,7 @@ class TestSwift3MultiDelete(Swift3TestCase):
req = Request.blank('/bucket?delete',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'Content-MD5': content_md5},
body=body)
req.date = datetime.now()
@ -93,6 +95,7 @@ class TestSwift3MultiDelete(Swift3TestCase):
req = Request.blank('/bucket?delete',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'Content-MD5': content_md5},
body=body)
status, headers, body = self.call_swift3(req)
@ -119,6 +122,7 @@ class TestSwift3MultiDelete(Swift3TestCase):
req = Request.blank('/bucket?delete',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'Content-MD5': content_md5},
body=body)
status, headers, body = self.call_swift3(req)
@ -135,6 +139,7 @@ class TestSwift3MultiDelete(Swift3TestCase):
req = Request.blank('/bucket?delete',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'Content-MD5': 'XXXX'},
body=body)
status, headers, body = self.call_swift3(req)
@ -150,7 +155,8 @@ class TestSwift3MultiDelete(Swift3TestCase):
req = Request.blank('/bucket?delete',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body=body)
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidRequest')
@ -167,6 +173,7 @@ class TestSwift3MultiDelete(Swift3TestCase):
req = Request.blank('/bucket?delete',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'Content-MD5': content_md5},
body=body)
status, headers, body = self.call_swift3(req)
@ -191,6 +198,7 @@ class TestSwift3MultiDelete(Swift3TestCase):
req = Request.blank('/bucket?delete',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS %s:hmac' % account,
'Date': self.get_date_header(),
'Content-MD5': content_md5},
body=body)
req.date = datetime.now()

126
swift3/test/unit/test_multi_upload.py
View File

@ -126,7 +126,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_bucket_upload_part(self):
req = Request.blank('/bucket?partNumber=1&uploadId=x',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidRequest')
@ -134,7 +135,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_object_multipart_uploads_list(self):
req = Request.blank('/bucket/object?uploads',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidRequest')
@ -142,7 +144,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_bucket_multipart_uploads_initiate(self):
req = Request.blank('/bucket?uploads',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidRequest')
@ -150,7 +153,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_bucket_list_parts(self):
req = Request.blank('/bucket?uploadId=x',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidRequest')
@ -158,7 +162,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_bucket_multipart_uploads_abort(self):
req = Request.blank('/bucket?uploadId=x',
environ={'REQUEST_METHOD': 'DELETE'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidRequest')
@ -166,7 +171,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_bucket_multipart_uploads_complete(self):
req = Request.blank('/bucket?uploadId=x',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidRequest')
@ -184,7 +190,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
query = '?uploads&' + query if query else '?uploads'
req = Request.blank('/bucket/%s' % query,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
return self.call_swift3(req)
@s3acl
@ -218,7 +225,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
req = Request.blank('/bucket?uploads',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, haeaders, body = self.call_swift3(req)
@ -239,7 +247,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
swob.HTTPNotFound, {}, '')
req = Request.blank('/bucket?uploads',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, haeaders, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '404')
self.assertEquals(self._get_error_code(body), 'NoSuchBucket')
@ -540,6 +549,7 @@ class TestSwift3MultiUpload(Swift3TestCase):
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization':
'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-meta-foo': 'bar'})
status, headers, body = self.call_swift3(req)
fromstring(body, 'InitiateMultipartUploadResult')
@ -555,6 +565,7 @@ class TestSwift3MultiUpload(Swift3TestCase):
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization':
'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-acl': 'public-read',
'x-amz-meta-foo': 'bar'})
status, headers, body = self.call_swift3(req)
@ -578,7 +589,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
req = Request.blank('/bucket/object?uploads',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization':
'AWS test:tester:hmac'})
'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '404')
self.assertEquals(self._get_error_code(body), 'NoSuchBucket')
@ -588,7 +600,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
malformed_xml = 'malformed_XML'
req = Request.blank('/bucket/object?uploadId=X',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body=malformed_xml)
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'MalformedXML')
@ -596,7 +609,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
# without target bucket
req = Request.blank('/nobucket/object?uploadId=X',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(), },
body=xml)
with patch('swift3.request.get_container_info',
lambda x, y: {'status': 404}):
@ -608,7 +622,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_object_multipart_upload_complete(self):
req = Request.blank('/bucket/object?uploadId=X',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(), },
body=xml)
status, headers, body = self.call_swift3(req)
fromstring(body, 'CompleteMultipartUploadResult')
@ -631,7 +646,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
swob.HTTPOk, headers, None)
req = Request.blank('/bucket/object?uploadId=X',
environ={'REQUEST_METHOD': 'POST'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body=xml)
status, headers, body = self.call_swift3(req)
fromstring(body, 'CompleteMultipartUploadResult')
@ -648,14 +664,16 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_object_multipart_upload_abort_error(self):
req = Request.blank('/bucket/object?uploadId=invalid',
environ={'REQUEST_METHOD': 'DELETE'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'NoSuchUpload')
# without target bucket
req = Request.blank('/nobucket/object?uploadId=X',
environ={'REQUEST_METHOD': 'DELETE'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
with patch('swift3.request.get_container_info',
lambda x, y: {'status': 404}):
self.swift.register('HEAD', '/v1/AUTH_test/nobucket',
@ -667,7 +685,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_object_multipart_upload_abort(self):
req = Request.blank('/bucket/object?uploadId=X',
environ={'REQUEST_METHOD': 'DELETE'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '204')
@ -677,7 +696,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
# without upload id
req = Request.blank('/bucket/object?partNumber=1',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body='part object')
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -685,7 +705,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
# invalid part number
req = Request.blank('/bucket/object?partNumber=invalid&uploadId=X',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body='part object')
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -693,7 +714,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
# part number must be > 0
req = Request.blank('/bucket/object?partNumber=0&uploadId=X',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body='part object')
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -701,7 +723,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
# part number must be < 1000
req = Request.blank('/bucket/object?partNumber=1001&uploadId=X',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body='part object')
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -709,7 +732,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
# without target bucket
req = Request.blank('/nobucket/object?partNumber=1&uploadId=X',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body='part object')
with patch('swift3.request.get_container_info',
lambda x, y: {'status': 404}):
@ -722,7 +746,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_object_upload_part(self):
req = Request.blank('/bucket/object?partNumber=1&uploadId=X',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body='part object')
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -731,14 +756,16 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_object_list_parts_error(self):
req = Request.blank('/bucket/object?uploadId=invalid',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'NoSuchUpload')
# without target bucket
req = Request.blank('/nobucket/object?uploadId=X',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
with patch('swift3.request.get_container_info',
lambda x, y: {'status': 404}):
self.swift.register('HEAD', '/v1/AUTH_test/nobucket',
@ -750,7 +777,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_object_list_parts(self):
req = Request.blank('/bucket/object?uploadId=X',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListPartsResult')
self.assertEquals(elem.find('Bucket').text, 'bucket')
@ -782,7 +810,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
swob.HTTPOk, {}, None)
req = Request.blank('/bucket/object@@?uploadId=X&encoding-type=url',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListPartsResult')
self.assertEquals(elem.find('Key').text, quote('object@@'))
@ -794,7 +823,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
swob.HTTPOk, {}, None)
req = Request.blank('/bucket/object@@?uploadId=X',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListPartsResult')
self.assertEquals(elem.find('Key').text, 'object@@')
@ -803,14 +833,16 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_object_list_parts_encoding_type_error(self):
req = Request.blank('/bucket/object?uploadId=X&encoding-type=xml',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
def test_object_list_parts_max_parts(self):
req = Request.blank('/bucket/object?uploadId=X&max-parts=1',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListPartsResult')
self.assertEquals(elem.find('IsTruncated').text, 'true')
@ -820,14 +852,16 @@ class TestSwift3MultiUpload(Swift3TestCase):
def test_object_list_parts_str_max_parts(self):
req = Request.blank('/bucket/object?uploadId=X&max-parts=invalid',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
def test_object_list_parts_negative_max_parts(self):
req = Request.blank('/bucket/object?uploadId=X&max-parts=-1',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -835,7 +869,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
req = Request.blank('/bucket/object?uploadId=X&max-parts=%d' %
(CONF.max_parts_listing + 1),
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListPartsResult')
self.assertEquals(elem.find('Bucket').text, 'bucket')
@ -864,7 +899,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
req = Request.blank('/bucket/object?uploadId=X&max-parts=%d' %
(MAX_32BIT_INT + 1),
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -872,7 +908,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
req = Request.blank('/bucket/object?uploadId=X&'
'part-number-marker=1',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListPartsResult')
self.assertEquals(len(elem.findall('Part')), 1)
@ -884,7 +921,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
req = Request.blank('/bucket/object?uploadId=X&part-number-marker='
'invalid',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -892,7 +930,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
req = Request.blank('/bucket/object?uploadId=X&part-number-marker='
'-1',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -901,7 +940,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
req = Request.blank('/bucket/object?uploadId=X&'
'part-number-marker=%s' % part_number_marker,
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListPartsResult')
self.assertEquals(len(elem.findall('Part')), 0)
@ -913,14 +953,16 @@ class TestSwift3MultiUpload(Swift3TestCase):
req = Request.blank('/bucket/object?uploadId=X&part-number-marker='
'%s' % ((MAX_32BIT_INT + 1)),
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
def test_object_list_parts_same_max_marts_as_objects_num(self):
req = Request.blank('/bucket/object?uploadId=X&max-parts=2',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
elem = fromstring(body, 'ListPartsResult')
self.assertEquals(len(elem.findall('Part')), 2)
@ -930,7 +972,8 @@ class TestSwift3MultiUpload(Swift3TestCase):
path = '/bucket%s' % ('/object' + query if hasObj else query)
req = Request.blank(path,
environ={'REQUEST_METHOD': method},
headers={'Authorization': 'AWS %s:hmac' % account},
headers={'Authorization': 'AWS %s:hmac' % account,
'Date': self.get_date_header()},
body=body)
return self.call_swift3(req)
@ -1066,6 +1109,7 @@ class TestSwift3MultiUpload(Swift3TestCase):
head_resp, src_o_headers, None)
put_headers = {'Authorization': 'AWS %s:hmac' % account,
'Date': self.get_date_header(),
'X-Amz-Copy-Source': src_path}
put_headers.update(put_header)
req = Request.blank(

54
swift3/test/unit/test_obj.py
View File

@ -82,7 +82,8 @@ class TestSwift3Obj(Swift3TestCase):
def _test_object_GETorHEAD(self, method):
req = Request.blank('/bucket/object',
environ={'REQUEST_METHOD': method},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -120,7 +121,8 @@ class TestSwift3Obj(Swift3TestCase):
# So, check the response code for error test of HEAD.
req = Request.blank('/bucket/object',
environ={'REQUEST_METHOD': 'HEAD'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
self.swift.register('HEAD', '/v1/AUTH_test/bucket/object',
swob.HTTPUnauthorized, {}, None)
status, headers, body = self.call_swift3(req)
@ -159,7 +161,8 @@ class TestSwift3Obj(Swift3TestCase):
req = Request.blank('/bucket/object',
environ={'REQUEST_METHOD': 'HEAD'},
headers={'Authorization': 'AWS test:tester:hmac',
'Range': range_value})
'Range': range_value,
'Date': self.get_date_header()})
return self.call_swift3(req)
@s3acl
@ -292,7 +295,8 @@ class TestSwift3Obj(Swift3TestCase):
req = Request.blank('/bucket/object',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac',
'Range': 'bytes=0-3'})
'Range': 'bytes=0-3',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '206')
@ -321,7 +325,8 @@ class TestSwift3Obj(Swift3TestCase):
'no-cache',
'attachment',
'gzip')},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -398,7 +403,8 @@ class TestSwift3Obj(Swift3TestCase):
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'x-amz-storage-class': 'STANDARD',
'Content-MD5': content_md5},
'Content-MD5': content_md5,
'Date': self.get_date_header()},
body=self.object_body)
req.date = datetime.now()
req.content_type = 'text/plain'
@ -424,7 +430,8 @@ class TestSwift3Obj(Swift3TestCase):
'X-Amz-Storage-Class': 'STANDARD',
'X-Amz-Meta-Something': 'oh hai',
'X-Amz-Copy-Source': '/some/source',
'Content-MD5': content_md5})
'Content-MD5': content_md5,
'Date': self.get_date_header()})
req.date = datetime.now()
req.content_type = 'text/plain'
status, headers, body = self.call_swift3(req)
@ -452,7 +459,8 @@ class TestSwift3Obj(Swift3TestCase):
head_resp, head_headers, None)
put_headers = {'Authorization': 'AWS test:tester:hmac',
'X-Amz-Copy-Source': '/some/source'}
'X-Amz-Copy-Source': '/some/source',
'Date': self.get_date_header()}
put_headers.update(put_header)
req = Request.blank('/bucket/object',
@ -486,7 +494,8 @@ class TestSwift3Obj(Swift3TestCase):
etag = '7dfa07a8e59ddbcd1dc84d4c4f82aea1'
last_modified_since = 'Fri, 01 Apr 2014 12:00:00 GMT'
header = {'X-Amz-Copy-Source-If-Match': etag}
header = {'X-Amz-Copy-Source-If-Match': etag,
'Date': self.get_date_header()}
status, header, body = \
self._test_object_PUT_copy(swob.HTTPPreconditionFailed,
header)
@ -516,7 +525,8 @@ class TestSwift3Obj(Swift3TestCase):
last_modified_since = 'Fri, 01 Apr 2014 11:00:00 GMT'
header = {'X-Amz-Copy-Source-If-Match': etag,
'X-Amz-Copy-Source-If-Modified-Since': last_modified_since}
'X-Amz-Copy-Source-If-Modified-Since': last_modified_since,
'Date': self.get_date_header()}
status, header, body = \
self._test_object_PUT_copy(swob.HTTPOk, header)
self.assertEquals(status.split()[0], '200')
@ -534,7 +544,8 @@ class TestSwift3Obj(Swift3TestCase):
last_modified_since = 'Fri, 01 Apr 2014 11:00:00 GMT'
header = {'X-Amz-Copy-Source-If-Match': etag,
'X-Amz-Copy-Source-If-Modified-Since': last_modified_since}
'X-Amz-Copy-Source-If-Modified-Since': last_modified_since,
'Date': self.get_date_header()}
status, header, body = \
self._test_object_PUT_copy(swob.HTTPOk, header)
@ -557,7 +568,8 @@ class TestSwift3Obj(Swift3TestCase):
last_modified_since = 'Fri, 01 Apr 2014 12:00:00 GMT'
header = {'X-Amz-Copy-Source-If-None-Match': etag,
'X-Amz-Copy-Source-If-Unmodified-Since': last_modified_since}
'X-Amz-Copy-Source-If-Unmodified-Since': last_modified_since,
'Date': self.get_date_header()}
status, header, body = \
self._test_object_PUT_copy(swob.HTTPOk, header)
@ -576,7 +588,8 @@ class TestSwift3Obj(Swift3TestCase):
last_modified_since = 'Fri, 01 Apr 2014 12:00:00 GMT'
header = {'X-Amz-Copy-Source-If-None-Match': etag,
'X-Amz-Copy-Source-If-Unmodified-Since': last_modified_since}
'X-Amz-Copy-Source-If-Unmodified-Since': last_modified_since,
'Date': self.get_date_header()}
status, header, body = \
self._test_object_PUT_copy(swob.HTTPOk, header)
self.assertEquals(status.split()[0], '200')
@ -627,7 +640,8 @@ class TestSwift3Obj(Swift3TestCase):
def test_object_DELETE_no_multipart(self):
req = Request.blank('/bucket/object',
environ={'REQUEST_METHOD': 'DELETE'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '204')
@ -642,7 +656,8 @@ class TestSwift3Obj(Swift3TestCase):
def test_object_DELETE_multipart(self):
req = Request.blank('/bucket/object',
environ={'REQUEST_METHOD': 'DELETE'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '204')
@ -663,7 +678,8 @@ class TestSwift3Obj(Swift3TestCase):
swob.HTTPOk, {}, '<SLO delete results>')
req = Request.blank('/bucket/object',
environ={'REQUEST_METHOD': 'DELETE'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEqual(status.split()[0], '204')
self.assertEqual(body, '')
@ -684,7 +700,8 @@ class TestSwift3Obj(Swift3TestCase):
def _test_object_for_s3acl(self, method, account):
req = Request.blank('/bucket/object',
environ={'REQUEST_METHOD': method},
headers={'Authorization': 'AWS %s:hmac' % account})
headers={'Authorization': 'AWS %s:hmac' % account,
'Date': self.get_date_header()})
return self.call_swift3(req)
def _test_set_container_permission(self, account, permission):
@ -785,7 +802,8 @@ class TestSwift3Obj(Swift3TestCase):
environ={'REQUEST_METHOD': 'PUT',
'HTTP_X_TIMESTAMP': '1396353600.000000'},
headers={'Authorization': 'AWS %s:hmac' % account,
'X-Amz-Copy-Source': src_path})
'X-Amz-Copy-Source': src_path,
'Date': self.get_date_header()})
return self.call_swift3(req)

70
swift3/test/unit/test_request.py
View File

@ -17,6 +17,7 @@ from contextlib import nested
from mock import patch, MagicMock
import unittest
from swift.common import swob
from swift.common.swob import Request, HTTPNoContent
from swift3.subresource import ACL, User, Owner, Grant, encode_acl
@ -97,7 +98,8 @@ class TestRequest(Swift3TestCase):
path = '/' + container + ('/' + obj if obj else '')
req = Request.blank(path,
environ={'REQUEST_METHOD': method},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
if issubclass(req_klass, S3AclRequest):
s3_req = req_klass(req.environ, MagicMock())
else:
@ -177,7 +179,8 @@ class TestRequest(Swift3TestCase):
req = Request.blank(
'/bucket?%s=%s' % (param, value),
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
return S3_Request(req.environ, True)
s3req = create_s3request_with_param('max-keys', '1')
@ -219,7 +222,8 @@ class TestRequest(Swift3TestCase):
def test_authenticate_delete_Authorization_from_s3req_headers(self):
req = Request.blank('/bucket/obj',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
with nested(patch.object(Request, 'get_response'),
patch.object(Request, 'remote_user', 'authorized')) \
as (m_swift_resp, m_remote_user):
@ -236,7 +240,8 @@ class TestRequest(Swift3TestCase):
method = 'GET'
req = Request.blank('/%s/%s' % (container, obj),
environ={'REQUEST_METHOD': method},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
with nested(patch.object(Request, 'get_response'),
patch.object(Request, 'remote_user', 'authorized')) \
as (m_swift_resp, m_remote_user):
@ -257,7 +262,8 @@ class TestRequest(Swift3TestCase):
req = Request.blank('/%s/%s' % (container, obj),
environ={'REQUEST_METHOD': method,
'swift.proxy_access_log_made': True},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
with nested(patch.object(Request, 'get_response'),
patch.object(Request, 'remote_user', 'authorized'),
patch('swift3.cfg.CONF.force_swift_request_proxy_log',
@ -273,7 +279,8 @@ class TestRequest(Swift3TestCase):
req = Request.blank('/%s/%s' % (container, obj),
environ={'REQUEST_METHOD': method,
'swift.proxy_access_log_made': True},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
with nested(patch.object(Request, 'get_response'),
patch.object(Request, 'remote_user', 'authorized')) \
as (m_swift_resp, m_remote_user):
@ -289,7 +296,8 @@ class TestRequest(Swift3TestCase):
'X-container-object-count': 5,
'X-container-meta-foo': 'bar'}, None)
req = Request.blank('/bucket', environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
s3_req = S3_Request(req.environ, True)
# first, call get_response('HEAD')
info = s3_req.get_container_info(self.app)
@ -314,6 +322,54 @@ class TestRequest(Swift3TestCase):
self.assertRaises(
expected_error, s3_req.get_container_info, MagicMock())
def test_date_header_missing(self):
self.swift.register('HEAD', '/v1/AUTH_test/nojunk', swob.HTTPNotFound,
{}, None)
req = Request.blank('/nojunk',
environ={'REQUEST_METHOD': 'HEAD'},
headers={'Authorization': 'AWS test:tester:hmac'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '403')
self.assertEquals(body, '')
def test_date_header_expired(self):
self.swift.register('HEAD', '/v1/AUTH_test/nojunk', swob.HTTPNotFound,
{}, None)
req = Request.blank('/nojunk',
environ={'REQUEST_METHOD': 'HEAD'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': 'Fri, 01 Apr 2014 12:00:00 GMT'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '403')
self.assertEquals(body, '')
def test_date_header_with_x_amz_date_valid(self):
self.swift.register('HEAD', '/v1/AUTH_test/nojunk', swob.HTTPNotFound,
{}, None)
req = Request.blank('/nojunk',
environ={'REQUEST_METHOD': 'HEAD'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': 'Fri, 01 Apr 2014 12:00:00 GMT',
'x-amz-date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '404')
self.assertEquals(body, '')
def test_date_header_with_x_amz_date_expired(self):
self.swift.register('HEAD', '/v1/AUTH_test/nojunk', swob.HTTPNotFound,
{}, None)
req = Request.blank('/nojunk',
environ={'REQUEST_METHOD': 'HEAD'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-date':
'Fri, 01 Apr 2014 12:00:00 GMT'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '403')
self.assertEquals(body, '')
if __name__ == '__main__':
unittest.main()

50
swift3/test/unit/test_s3_acl.py
View File

@ -168,7 +168,8 @@ class TestSwift3S3Acl(Swift3TestCase):
def test_bucket_acl_PUT_with_other_owner(self):
req = Request.blank('/bucket?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body=tostring(
ACLPrivate(
Owner(id='test:other',
@ -179,7 +180,8 @@ class TestSwift3S3Acl(Swift3TestCase):
def test_object_acl_PUT_xml_error(self):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body="invalid xml")
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'MalformedACLError')
@ -188,6 +190,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-acl': 'private'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -196,6 +199,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-acl': 'public-read'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -204,6 +208,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-acl': 'public-read-write'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -212,6 +217,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-acl': 'authenticated-read'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -220,6 +226,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-acl': 'bucket-owner-read'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -228,6 +235,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-acl': 'bucket-owner-full-control'})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -236,6 +244,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-acl': 'invalid'})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -244,6 +253,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-grant-' + permission:
'id=test:tester'})
return self.call_swift3(req)
@ -276,6 +286,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-grant-full-control':
'id=test:tester'},
body=tostring(
@ -289,6 +300,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-grant-full-control':
'id=test:tester',
'x-amz-acl': 'public-read'})
@ -299,6 +311,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-grant-read': 'emailAddress=a@b.c'})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'NotImplemented')
@ -310,7 +323,8 @@ class TestSwift3S3Acl(Swift3TestCase):
xml = _make_xml(grantee=grantee)
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body=xml)
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'NotImplemented')
@ -321,7 +335,8 @@ class TestSwift3S3Acl(Swift3TestCase):
xml = _make_xml(grantee=grantee)
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body=xml)
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'MalformedACLError')
@ -330,6 +345,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-grant-read':
'uri="http://acs.amazonaws.com/groups/'
'global/AuthenticatedUsers"'})
@ -340,6 +356,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-grant-read':
'uri="http://acs.amazonaws.com/groups/'
'global/AllUsers"'})
@ -350,6 +367,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-grant-read':
'uri="http://localhost/"'})
status, headers, body = self.call_swift3(req)
@ -363,7 +381,8 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()},
body=xml)
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -372,6 +391,7 @@ class TestSwift3S3Acl(Swift3TestCase):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header(),
'x-amz-grant-read': 'key=value'})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'InvalidArgument')
@ -379,7 +399,8 @@ class TestSwift3S3Acl(Swift3TestCase):
def _test_bucket_acl_GET(self, account):
req = Request.blank('/bucket?acl',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS %s:hmac' % account})
headers={'Authorization': 'AWS %s:hmac' % account,
'Date': self.get_date_header()})
return self.call_swift3(req)
def test_bucket_acl_GET_without_permission(self):
@ -402,7 +423,8 @@ class TestSwift3S3Acl(Swift3TestCase):
acl = ACL(self.default_owner, [Grant(User(account), permission)])
req = Request.blank('/bucket?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS %s:hmac' % account},
headers={'Authorization': 'AWS %s:hmac' % account,
'Date': self.get_date_header()},
body=tostring(acl.elem()))
return self.call_swift3(req)
@ -426,7 +448,8 @@ class TestSwift3S3Acl(Swift3TestCase):
def _test_object_acl_GET(self, account):
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS %s:hmac' % account})
headers={'Authorization': 'AWS %s:hmac' % account,
'Date': self.get_date_header()})
return self.call_swift3(req)
def test_object_acl_GET_without_permission(self):
@ -449,7 +472,8 @@ class TestSwift3S3Acl(Swift3TestCase):
acl = ACL(self.default_owner, [Grant(User(account), permission)])
req = Request.blank('/bucket/object?acl',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS %s:hmac' % account},
headers={'Authorization': 'AWS %s:hmac' % account,
'Date': self.get_date_header()},
body=tostring(acl.elem()))
return self.call_swift3(req)
@ -473,7 +497,7 @@ class TestSwift3S3Acl(Swift3TestCase):
def test_s3acl_decorator(self):
@s3acl
def non_class_s3acl_error():
raise
raise TypeError()
class FakeClass(object):
def __init__(self):
@ -481,7 +505,7 @@ class TestSwift3S3Acl(Swift3TestCase):
@s3acl
def s3acl_error(self):
raise
raise TypeError()
@s3acl
def s3acl_assert_fail(self):
@ -490,12 +514,12 @@ class TestSwift3S3Acl(Swift3TestCase):
@s3acl(s3acl_only=True)
def s3acl_s3only_error(self):
if CONF.s3_acl:
raise
raise TypeError()
@s3acl(s3acl_only=True)
def s3acl_s3only_no_error(self):
if not CONF.s3_acl:
raise
raise TypeError()
fake_class = FakeClass()

12
swift3/test/unit/test_service.py
View File

@ -62,7 +62,8 @@ class TestSwift3Service(Swift3TestCase):
def test_service_GET(self):
req = Request.blank('/',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -85,7 +86,8 @@ class TestSwift3Service(Swift3TestCase):
def test_service_GET_subresource(self):
req = Request.blank('/?acl',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -114,7 +116,8 @@ class TestSwift3Service(Swift3TestCase):
req = Request.blank('/',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
@ -142,7 +145,8 @@ class TestSwift3Service(Swift3TestCase):
req = Request.blank('/',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
return self.call_swift3(req)
@s3acl(s3acl_only=True)

10
swift3/test/unit/test_versioning.py
View File

@ -29,7 +29,9 @@ class TestSwift3Versioning(Swift3TestCase):
def test_object_versioning_GET(self):
req = Request.blank('/bucket/object?versioning',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(status.split()[0], '200')
fromstring(body, 'VersioningConfiguration')
@ -37,14 +39,16 @@ class TestSwift3Versioning(Swift3TestCase):
def test_object_versioning_PUT(self):
req = Request.blank('/bucket/object?versioning',
environ={'REQUEST_METHOD': 'PUT'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
self.assertEquals(self._get_error_code(body), 'NotImplemented')
def test_bucket_versioning_GET(self):
req = Request.blank('/bucket?versioning',
environ={'REQUEST_METHOD': 'GET'},
headers={'Authorization': 'AWS test:tester:hmac'})
headers={'Authorization': 'AWS test:tester:hmac',
'Date': self.get_date_header()})
status, headers, body = self.call_swift3(req)
fromstring(body, 'VersioningConfiguration')

2
test-requirements.txt
View File

@ -5,5 +5,5 @@ openstack.nose_plugin
coverage
mock
pylint
python-openstackclient
python-openstackclient<=1.9.0
boto