diff --git a/test/functional/swift_test_client.py b/test/functional/swift_test_client.py index 750181bc06..0148ba7b33 100644 --- a/test/functional/swift_test_client.py +++ b/test/functional/swift_test_client.py @@ -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 diff --git a/test/functional/tests.py b/test/functional/tests.py index 758de80802..092c8098f6 100644 --- a/test/functional/tests.py +++ b/test/functional/tests.py @@ -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( diff --git a/test/unit/common/middleware/test_tempurl.py b/test/unit/common/middleware/test_tempurl.py index c84063120a..00f0af1d71 100644 --- a/test/unit/common/middleware/test_tempurl.py +++ b/test/unit/common/middleware/test_tempurl.py @@ -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({