Add test that a tempurl POST cannot set a DLO manifest header

Follow up to [1] to add tests for tempurl POSTs not being allowed
to set a DLO manifest header.

[1] I11e68830009d3f6bff44ae4011a41b67139146f6

Change-Id: I7c0ad5a936f71e56c599b8495a586913d3334422
Related-Bug: 1453948
This commit is contained in:
Alistair Coles 2015-08-26 16:30:23 +01:00
parent d4409c0a04
commit 58a10a5fff
3 changed files with 52 additions and 13 deletions

View File

@ -1018,3 +1018,26 @@ class File(Base):
raise ResponseError(self.conn.response)
self.md5 = self.compute_md5sum(six.StringIO(data))
return resp
def post(self, hdrs=None, parms=None, cfg=None, return_resp=False):
if hdrs is None:
hdrs = {}
if parms is None:
parms = {}
if cfg is None:
cfg = {}
headers = self.make_headers(cfg=cfg)
headers.update(hdrs)
self.conn.make_request('POST', self.path, hdrs=headers,
parms=parms, cfg=cfg)
if self.conn.response.status not in (201, 202):
raise ResponseError(self.conn.response, 'POST',
self.conn.make_path(self.path))
if return_resp:
return self.conn.response
return True

View File

@ -3197,6 +3197,22 @@ class TestTempurl(Base):
else:
self.fail('request did not error')
# try again using a tempurl POST to an already created object
new_obj.write('', {}, parms=put_parms, cfg={'no_auth_token': True})
expires = int(time.time()) + 86400
sig = self.tempurl_sig(
'POST', expires, self.env.conn.make_path(new_obj.path),
self.env.tempurl_key)
post_parms = {'temp_url_sig': sig,
'temp_url_expires': str(expires)}
try:
new_obj.post({'x-object-manifest': '%s/foo' % other_container},
parms=post_parms, cfg={'no_auth_token': True})
except ResponseError as e:
self.assertEqual(e.status, 400)
else:
self.fail('request did not error')
def test_HEAD(self):
expires = int(time.time()) + 86400
sig = self.tempurl_sig(

View File

@ -737,22 +737,22 @@ class TestTempURL(unittest.TestCase):
def test_disallowed_header_object_manifest(self):
self.tempurl = tempurl.filter_factory({})(self.auth)
method = 'PUT'
expires = int(time() + 86400)
path = '/v1/a/c/o'
key = 'abc'
hmac_body = '%s\n%s\n%s' % (method, expires, path)
sig = hmac.new(key, hmac_body, sha1).hexdigest()
req = self._make_request(
path, method='PUT', keys=[key],
headers={'x-object-manifest': 'private/secret'},
environ={'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
sig, expires)})
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 400)
self.assertTrue('header' in resp.body)
self.assertTrue('not allowed' in resp.body)
self.assertTrue('X-Object-Manifest' in resp.body)
for method in ('PUT', 'POST'):
hmac_body = '%s\n%s\n%s' % (method, expires, path)
sig = hmac.new(key, hmac_body, sha1).hexdigest()
req = self._make_request(
path, method=method, keys=[key],
headers={'x-object-manifest': 'private/secret'},
environ={'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s'
% (sig, expires)})
resp = req.get_response(self.tempurl)
self.assertEquals(resp.status_int, 400)
self.assertTrue('header' in resp.body)
self.assertTrue('not allowed' in resp.body)
self.assertTrue('X-Object-Manifest' in resp.body)
def test_removed_incoming_header(self):
self.tempurl = tempurl.filter_factory({