TempURL changes
* Changes to all tests that enable tests to run with sha1 and sha2 encryption when generating the signature. Change-Id: I890f30358f8ef1955ce3d247e6997d8461bfbd93
This commit is contained in:
parent
7aeff6cfb4
commit
bd4dffc5ba
|
@ -14,22 +14,35 @@ See the License for the specific language governing permissions and
|
||||||
limitations under the License.
|
limitations under the License.
|
||||||
"""
|
"""
|
||||||
import time
|
import time
|
||||||
|
|
||||||
from unittest import skipUnless
|
from unittest import skipUnless
|
||||||
|
|
||||||
|
from cafe.drivers.unittest.datasets import DatasetList
|
||||||
|
from cafe.drivers.unittest.decorators import (
|
||||||
|
DataDrivenFixture, data_driven_test)
|
||||||
from cafe.engine.http.client import HTTPClient
|
from cafe.engine.http.client import HTTPClient
|
||||||
from cloudcafe.common.tools.check_dict import get_value
|
from cloudcafe.common.tools.check_dict import get_value
|
||||||
from cloudcafe.objectstorage.objectstorage_api.common.constants import \
|
from cloudcafe.objectstorage.objectstorage_api.common.constants import \
|
||||||
Constants
|
Constants
|
||||||
from cloudroast.objectstorage.fixtures import ObjectStorageFixture
|
from cloudroast.objectstorage.fixtures import ObjectStorageFixture
|
||||||
|
|
||||||
|
|
||||||
BASE_CONTAINER_NAME = 'tempurl'
|
BASE_CONTAINER_NAME = 'tempurl'
|
||||||
CONTENT_TYPE_TEXT = 'text/plain; charset=UTF-8'
|
CONTENT_TYPE_TEXT = 'text/plain; charset=UTF-8'
|
||||||
TEMPURL_KEY_LIFE = 20
|
TEMPURL_KEY_LIFE = 20
|
||||||
EXPECTED_DISPOSITION = ("attachment; filename=\"{filename}\";"
|
EXPECTED_DISPOSITION = ("attachment; filename=\"{filename}\";"
|
||||||
" filename*=UTF-8\'\'{filename}")
|
" filename*=UTF-8\'\'{filename}")
|
||||||
|
|
||||||
|
sha_type = DatasetList()
|
||||||
|
|
||||||
|
sha_type.append_new_dataset(
|
||||||
|
"with_sha1",
|
||||||
|
{"sha_type": "sha1"})
|
||||||
|
sha_type.append_new_dataset(
|
||||||
|
"with_sha2",
|
||||||
|
{"sha_type": "sha2"})
|
||||||
|
|
||||||
|
|
||||||
|
@DataDrivenFixture
|
||||||
class TempUrlTest(ObjectStorageFixture):
|
class TempUrlTest(ObjectStorageFixture):
|
||||||
def _reset_default_key(self):
|
def _reset_default_key(self):
|
||||||
response = None
|
response = None
|
||||||
|
@ -57,7 +70,7 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
"""
|
"""
|
||||||
Check if the TempURL has been changed, if so, change it back to
|
Check if the TempURL has been changed, if so, change it back to
|
||||||
the expected key and wait the appropriate amount of time to let the
|
the expected key and wait the appropriate amount of time to let the
|
||||||
key propogate through the system.
|
key propagate through the system.
|
||||||
"""
|
"""
|
||||||
super(TempUrlTest, self).setUp()
|
super(TempUrlTest, self).setUp()
|
||||||
|
|
||||||
|
@ -66,24 +79,35 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
if not response.ok:
|
if not response.ok:
|
||||||
raise Exception('Could not set TempURL key.')
|
raise Exception('Could not set TempURL key.')
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_object_creation_via_tempurl(self):
|
def ddtest_object_creation_via_tempurl(self, sha_type=None):
|
||||||
"""
|
"""
|
||||||
Scenario:
|
Scenario:
|
||||||
Create a 'PUT' TempURL. Perform a HTTP PUT to the TempURL sending
|
Create a 'PUT' TempURL, using sha1 and sha2 encryption
|
||||||
data for the object;.
|
algorithms. Perform a HTTP PUT to the TempURL sending
|
||||||
|
data for the object.
|
||||||
|
|
||||||
Expected Results:
|
Expected Results:
|
||||||
The object should be created containing the correct data.
|
The object should be created containing the correct data.
|
||||||
"""
|
"""
|
||||||
container_name = self.create_temp_container(BASE_CONTAINER_NAME)
|
container_name = self.create_temp_container(BASE_CONTAINER_NAME)
|
||||||
|
|
||||||
tempurl_data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'PUT',
|
tempurl_data = self.client.create_temp_url(
|
||||||
container_name,
|
'PUT',
|
||||||
self.object_name,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.object_name,
|
||||||
self.tempurl_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'PUT',
|
||||||
|
container_name,
|
||||||
|
self.object_name,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
|
@ -103,29 +127,37 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
'X-Object-Meta-Foo': 'bar'}
|
'X-Object-Meta-Foo': 'bar'}
|
||||||
params = {'temp_url_sig': tempurl_data.get('signature'),
|
params = {'temp_url_sig': tempurl_data.get('signature'),
|
||||||
'temp_url_expires': tempurl_data.get('expires')}
|
'temp_url_expires': tempurl_data.get('expires')}
|
||||||
response = self.http.put(
|
put_response = self.http.put(
|
||||||
tempurl_data.get('target_url'),
|
tempurl_data.get('target_url'),
|
||||||
params=params,
|
params=params,
|
||||||
headers=headers,
|
headers=headers,
|
||||||
data=self.object_data)
|
data=self.object_data)
|
||||||
|
|
||||||
response = self.client.get_object(container_name, self.object_name)
|
self.assertEqual(put_response.status_code,
|
||||||
|
201,
|
||||||
|
msg='Failed to PUT object '
|
||||||
|
'with status code {0}'.format(
|
||||||
|
put_response.status_code))
|
||||||
|
|
||||||
|
object_response = self.client.get_object(container_name,
|
||||||
|
self.object_name)
|
||||||
|
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
response.content,
|
object_response.content,
|
||||||
self.object_data,
|
self.object_data,
|
||||||
msg="object should contain correct data")
|
msg="object should contain correct data")
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'x-object-meta-foo',
|
'x-object-meta-foo',
|
||||||
response.headers,
|
object_response.headers,
|
||||||
msg="x-object-meta-foo header was not set")
|
msg="x-object-meta-foo header was not set")
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
response.headers['x-object-meta-foo'],
|
object_response.headers['x-object-meta-foo'],
|
||||||
'bar',
|
'bar',
|
||||||
msg="x-object-meta-foo header value is not bar")
|
msg="x-object-meta-foo header value is not bar")
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_object_retrieval_via_tempurl(self):
|
def ddtest_object_retrieval_via_tempurl(self, sha_type=None):
|
||||||
"""
|
"""
|
||||||
Scenario:
|
Scenario:
|
||||||
Create a 'GET' TempURL for an existing object. Perform a HTTP GET
|
Create a 'GET' TempURL for an existing object. Perform a HTTP GET
|
||||||
|
@ -141,12 +173,21 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
self.object_name,
|
self.object_name,
|
||||||
data=self.object_data)
|
data=self.object_data)
|
||||||
|
|
||||||
tempurl_data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'GET',
|
tempurl_data = self.client.create_temp_url(
|
||||||
container_name,
|
'GET',
|
||||||
self.object_name,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.object_name,
|
||||||
self.tempurl_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'GET',
|
||||||
|
container_name,
|
||||||
|
self.object_name,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
|
@ -187,8 +228,10 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
self.object_data,
|
self.object_data,
|
||||||
'object should contain correct data.')
|
'object should contain correct data.')
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_object_retrieval_via_tempurl_content_disposition_inline(self):
|
def ddtest_object_retrieval_via_tempurl_content_disposition_inline(
|
||||||
|
self, sha_type=None):
|
||||||
"""
|
"""
|
||||||
Scenario:
|
Scenario:
|
||||||
Create a 'GET' TempURL for an existing object. Perform a HTTP GET
|
Create a 'GET' TempURL for an existing object. Perform a HTTP GET
|
||||||
|
@ -207,12 +250,21 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
headers={'content-disposition': attachment_hdr},
|
headers={'content-disposition': attachment_hdr},
|
||||||
data=self.object_data)
|
data=self.object_data)
|
||||||
|
|
||||||
tempurl_data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'GET',
|
tempurl_data = self.client.create_temp_url(
|
||||||
container_name,
|
'GET',
|
||||||
self.object_name,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.object_name,
|
||||||
self.tempurl_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'GET',
|
||||||
|
container_name,
|
||||||
|
self.object_name,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
|
@ -251,8 +303,10 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
self.object_data,
|
self.object_data,
|
||||||
'object should contain correct data.')
|
'object should contain correct data.')
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_tempurl_content_disposition_filename_with_trailing_slash(self):
|
def ddtest_tempurl_content_disposition_filename_with_trailing_slash(
|
||||||
|
self, sha_type=None):
|
||||||
"""
|
"""
|
||||||
Scenario:
|
Scenario:
|
||||||
Create a 'GET' TempURL for an object where the object name ends
|
Create a 'GET' TempURL for an object where the object name ends
|
||||||
|
@ -277,30 +331,39 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
|
|
||||||
self.assertTrue(set_key_response.ok, msg="tempurl key was not set")
|
self.assertTrue(set_key_response.ok, msg="tempurl key was not set")
|
||||||
|
|
||||||
data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'GET',
|
tempurl_data = self.client.create_temp_url(
|
||||||
container_name,
|
'GET',
|
||||||
self.obj_name_containing_trailing_slash,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.obj_name_containing_trailing_slash,
|
||||||
self.tempurl_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'GET',
|
||||||
|
container_name,
|
||||||
|
self.obj_name_containing_trailing_slash,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='target_url was not in the created tempurl')
|
msg='target_url was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'signature',
|
'signature',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='signature was not in the created tempurl')
|
msg='signature was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'expires',
|
'expires',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='expires was not in the created tempurl')
|
msg='expires was not in the created tempurl')
|
||||||
|
|
||||||
params = {'temp_url_sig': data.get('signature'),
|
params = {'temp_url_sig': tempurl_data.get('signature'),
|
||||||
'temp_url_expires': data.get('expires')}
|
'temp_url_expires': tempurl_data.get('expires')}
|
||||||
tempurl_get_response = self.http.get(
|
tempurl_get_response = self.http.get(
|
||||||
data.get('target_url'), params=params)
|
tempurl_data.get('target_url'), params=params)
|
||||||
|
|
||||||
expected_filename = \
|
expected_filename = \
|
||||||
self.obj_name_containing_trailing_slash.split('/')[0]
|
self.obj_name_containing_trailing_slash.split('/')[0]
|
||||||
|
@ -325,9 +388,10 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
self.object_data,
|
self.object_data,
|
||||||
'object should contain correct data.')
|
'object should contain correct data.')
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_tempurl_content_disposition_header_filename_with_trailing_slash(
|
def ddtest_tempurl_content_disposition_header_filename_with_trailing_slash(
|
||||||
self):
|
self, sha_type=None):
|
||||||
"""
|
"""
|
||||||
Scenario:
|
Scenario:
|
||||||
Create a 'GET' TempURL for an object where the object name ends
|
Create a 'GET' TempURL for an object where the object name ends
|
||||||
|
@ -355,42 +419,51 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
|
|
||||||
self.assertTrue(set_key_response.ok, msg="tempurl key was not set")
|
self.assertTrue(set_key_response.ok, msg="tempurl key was not set")
|
||||||
|
|
||||||
data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'GET',
|
tempurl_data = self.client.create_temp_url(
|
||||||
container_name,
|
'GET',
|
||||||
self.obj_name_containing_trailing_slash,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.obj_name_containing_trailing_slash,
|
||||||
self.tempurl_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'GET',
|
||||||
|
container_name,
|
||||||
|
self.obj_name_containing_trailing_slash,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='target_url was not in the created tempurl')
|
msg='target_url was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'signature',
|
'signature',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='signature was not in the created tempurl')
|
msg='signature was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'expires',
|
'expires',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='expires was not in the created tempurl')
|
msg='expires was not in the created tempurl')
|
||||||
|
|
||||||
params = {'temp_url_sig': data.get('signature'),
|
params = {'temp_url_sig': tempurl_data.get('signature'),
|
||||||
'temp_url_expires': data.get('expires')}
|
'temp_url_expires': tempurl_data.get('expires')}
|
||||||
tempurl_get_response = self.http.get(
|
tempurl_get_response = self.http.get(
|
||||||
data.get('target_url'), params=params)
|
tempurl_data.get('target_url'), params=params)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='target_url was not in the created tempurl')
|
msg='target_url was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'signature',
|
'signature',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='signature was not in the created tempurl')
|
msg='signature was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'expires',
|
'expires',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='expires was not in the created tempurl')
|
msg='expires was not in the created tempurl')
|
||||||
|
|
||||||
expected_disposition = content_disposition
|
expected_disposition = content_disposition
|
||||||
|
@ -412,8 +485,10 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
self.object_data,
|
self.object_data,
|
||||||
'object should contain correct data.')
|
'object should contain correct data.')
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_tempurl_content_disposition_filename_containing_slash(self):
|
def ddtest_tempurl_content_disposition_filename_containing_slash(
|
||||||
|
self, sha_type=None):
|
||||||
"""
|
"""
|
||||||
Scenario:
|
Scenario:
|
||||||
Create a 'GET' TempURL for an object where the object name
|
Create a 'GET' TempURL for an object where the object name
|
||||||
|
@ -439,46 +514,54 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
|
|
||||||
self.assertTrue(set_key_response.ok, msg="tempurl key was not set")
|
self.assertTrue(set_key_response.ok, msg="tempurl key was not set")
|
||||||
|
|
||||||
data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'GET',
|
tempurl_data = self.client.create_temp_url(
|
||||||
container_name,
|
'GET',
|
||||||
self.obj_name_containing_slash,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.obj_name_containing_slash,
|
||||||
self.tempurl_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'GET',
|
||||||
|
container_name,
|
||||||
|
self.obj_name_containing_slash,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='target_url was not in the created tempurl')
|
msg='target_url was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'signature',
|
'signature',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='signature was not in the created tempurl')
|
msg='signature was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'expires',
|
'expires',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='expires was not in the created tempurl')
|
msg='expires was not in the created tempurl')
|
||||||
|
|
||||||
params = {'temp_url_sig': data.get('signature'),
|
params = {'temp_url_sig': tempurl_data.get('signature'),
|
||||||
'temp_url_expires': data.get('expires')}
|
'temp_url_expires': tempurl_data.get('expires')}
|
||||||
tempurl_get_response = self.http.get(
|
tempurl_get_response = self.http.get(
|
||||||
data.get('target_url'), params=params)
|
tempurl_data.get('target_url'), params=params)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='target_url was not in the created tempurl')
|
msg='target_url was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'signature',
|
'signature',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='signature was not in the created tempurl')
|
msg='signature was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'expires',
|
'expires',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='expires was not in the created tempurl')
|
msg='expires was not in the created tempurl')
|
||||||
|
|
||||||
expected_filename = \
|
expected_filename = self.obj_name_containing_slash.split('/')[1]
|
||||||
expected_filename = self.obj_name_containing_slash.split('/')[1]
|
|
||||||
|
|
||||||
expected_disposition = EXPECTED_DISPOSITION.format(
|
expected_disposition = EXPECTED_DISPOSITION.format(
|
||||||
filename=expected_filename)
|
filename=expected_filename)
|
||||||
|
@ -500,9 +583,10 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
self.object_data,
|
self.object_data,
|
||||||
'object should contain correct data.')
|
'object should contain correct data.')
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_tempurl_content_disposition_header_filename_containing_slash(
|
def ddtest_tempurl_content_disposition_header_filename_containing_slash(
|
||||||
self):
|
self, sha_type=None):
|
||||||
"""
|
"""
|
||||||
Scenario:
|
Scenario:
|
||||||
Create a 'GET' TempURL for an object where the object name
|
Create a 'GET' TempURL for an object where the object name
|
||||||
|
@ -530,42 +614,51 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
|
|
||||||
self.assertTrue(set_key_response.ok, msg="tempurl key was not set")
|
self.assertTrue(set_key_response.ok, msg="tempurl key was not set")
|
||||||
|
|
||||||
data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'GET',
|
tempurl_data = self.client.create_temp_url(
|
||||||
container_name,
|
'GET',
|
||||||
self.obj_name_containing_slash,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.obj_name_containing_slash,
|
||||||
self.tempurl_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'GET',
|
||||||
|
container_name,
|
||||||
|
self.obj_name_containing_slash,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='target_url was not in the created tempurl')
|
msg='target_url was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'signature',
|
'signature',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='signature was not in the created tempurl')
|
msg='signature was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'expires',
|
'expires',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='expires was not in the created tempurl')
|
msg='expires was not in the created tempurl')
|
||||||
|
|
||||||
params = {'temp_url_sig': data.get('signature'),
|
params = {'temp_url_sig': tempurl_data.get('signature'),
|
||||||
'temp_url_expires': data.get('expires')}
|
'temp_url_expires': tempurl_data.get('expires')}
|
||||||
tempurl_get_response = self.http.get(
|
tempurl_get_response = self.http.get(
|
||||||
data.get('target_url'), params=params)
|
tempurl_data.get('target_url'), params=params)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='target_url was not in the created tempurl')
|
msg='target_url was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'signature',
|
'signature',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='signature was not in the created tempurl')
|
msg='signature was not in the created tempurl')
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'expires',
|
'expires',
|
||||||
data.keys(),
|
tempurl_data.keys(),
|
||||||
msg='expires was not in the created tempurl')
|
msg='expires was not in the created tempurl')
|
||||||
|
|
||||||
expected_disposition = content_disposition
|
expected_disposition = content_disposition
|
||||||
|
@ -587,8 +680,10 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
self.object_data,
|
self.object_data,
|
||||||
'object should contain correct data.')
|
'object should contain correct data.')
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_object_retrieval_with_filename_override_param(self):
|
def ddtest_object_retrieval_with_filename_override_param(
|
||||||
|
self, sha_type=None):
|
||||||
"""
|
"""
|
||||||
Scenario:
|
Scenario:
|
||||||
Create a 'GET' TempURL. Download the object using the TempURL,
|
Create a 'GET' TempURL. Download the object using the TempURL,
|
||||||
|
@ -611,12 +706,21 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
headers=headers,
|
headers=headers,
|
||||||
data=self.object_data)
|
data=self.object_data)
|
||||||
|
|
||||||
tempurl_data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'GET',
|
tempurl_data = self.client.create_temp_url(
|
||||||
container_name,
|
'GET',
|
||||||
self.object_name,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.object_name,
|
||||||
self.tempurl_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'GET',
|
||||||
|
container_name,
|
||||||
|
self.object_name,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
|
@ -655,8 +759,10 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
self.object_data,
|
self.object_data,
|
||||||
'object should contain correct data.')
|
'object should contain correct data.')
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_filename_override_param_containing_trailing_slash(self):
|
def ddtest_filename_override_param_containing_trailing_slash(
|
||||||
|
self, sha_type=None):
|
||||||
"""
|
"""
|
||||||
Scenario:
|
Scenario:
|
||||||
Create a 'GET' TempURL. Download the object using the TempURL,
|
Create a 'GET' TempURL. Download the object using the TempURL,
|
||||||
|
@ -681,12 +787,21 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
headers=headers,
|
headers=headers,
|
||||||
data=self.object_data)
|
data=self.object_data)
|
||||||
|
|
||||||
tempurl_data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'GET',
|
tempurl_data = self.client.create_temp_url(
|
||||||
container_name,
|
'GET',
|
||||||
self.object_name,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.object_name,
|
||||||
self.tempurl_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'GET',
|
||||||
|
container_name,
|
||||||
|
self.object_name,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
|
@ -725,8 +840,9 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
self.object_data,
|
self.object_data,
|
||||||
'object should contain correct data.')
|
'object should contain correct data.')
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_filename_override_param_containing_slash(self):
|
def ddtest_filename_override_param_containing_slash(self, sha_type=None):
|
||||||
"""
|
"""
|
||||||
Scenario:
|
Scenario:
|
||||||
Create a 'GET' TempURL. Download the object using the TempURL,
|
Create a 'GET' TempURL. Download the object using the TempURL,
|
||||||
|
@ -751,12 +867,21 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
headers=headers,
|
headers=headers,
|
||||||
data=self.object_data)
|
data=self.object_data)
|
||||||
|
|
||||||
tempurl_data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'GET',
|
tempurl_data = self.client.create_temp_url(
|
||||||
container_name,
|
'GET',
|
||||||
self.object_name,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.object_name,
|
||||||
self.tempurl_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'GET',
|
||||||
|
container_name,
|
||||||
|
self.object_name,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
|
@ -795,10 +920,11 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
self.object_data,
|
self.object_data,
|
||||||
'object should contain correct data.')
|
'object should contain correct data.')
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
@skipUnless(get_value('tempurl-methods') == 'delete',
|
@skipUnless(get_value('tempurl-methods') == 'delete',
|
||||||
'tempurl must be configured to allow DELETE.')
|
'tempurl must be configured to allow DELETE.')
|
||||||
def test_tempurl_object_delete(self):
|
def ddtest_tempurl_object_delete(self, sha_type=None):
|
||||||
container_name = self.create_temp_container(BASE_CONTAINER_NAME)
|
container_name = self.create_temp_container(BASE_CONTAINER_NAME)
|
||||||
|
|
||||||
headers = {'Content-Length': self.content_length,
|
headers = {'Content-Length': self.content_length,
|
||||||
|
@ -809,12 +935,21 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
headers=headers,
|
headers=headers,
|
||||||
data=self.object_data)
|
data=self.object_data)
|
||||||
|
|
||||||
tempurl_data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'DELETE',
|
tempurl_data = self.client.create_temp_url(
|
||||||
container_name,
|
'DELETE',
|
||||||
self.object_name,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.object_name,
|
||||||
self.tempurl_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'DELETE',
|
||||||
|
container_name,
|
||||||
|
self.object_name,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
|
@ -831,22 +966,22 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
|
|
||||||
params = {'temp_url_sig': tempurl_data.get('signature'),
|
params = {'temp_url_sig': tempurl_data.get('signature'),
|
||||||
'temp_url_expires': tempurl_data.get('expires')}
|
'temp_url_expires': tempurl_data.get('expires')}
|
||||||
delete_response = self.http.delete(
|
delete_response = self.http.delete(tempurl_data.get('target_url'),
|
||||||
tempurl_data.get('target_url'),
|
params=params)
|
||||||
params=params)
|
|
||||||
|
|
||||||
self.assertEqual(delete_response.status_code, 204)
|
self.assertEqual(delete_response.status_code,
|
||||||
|
204,
|
||||||
|
msg='Failed to DELETE object with status'
|
||||||
|
' code {0}'.format(delete_response.status_code))
|
||||||
|
|
||||||
get_response = self.client.get_object(container_name, self.object_name)
|
get_response = self.client.get_object(container_name, self.object_name)
|
||||||
|
|
||||||
self.assertEqual(get_response.status_code, 404)
|
self.assertEqual(get_response.status_code, 404)
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@skipUnless(get_value('slow') == 'true', 'sleep for key change')
|
@skipUnless(get_value('slow') == 'true', 'sleep for key change')
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_object_retrieval_with_two_tempurl_keys(self):
|
def ddtest_object_retrieval_with_two_tempurl_keys(self, sha_type=None):
|
||||||
time.sleep(
|
|
||||||
float(self.objectstorage_api_config.tempurl_key_cache_time))
|
|
||||||
|
|
||||||
container_name = self.create_temp_container('temp_url')
|
container_name = self.create_temp_container('temp_url')
|
||||||
|
|
||||||
headers = {'Content-Length': self.content_length,
|
headers = {'Content-Length': self.content_length,
|
||||||
|
@ -870,12 +1005,21 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
|
|
||||||
self.assertTrue(set_bar_key_response.ok)
|
self.assertTrue(set_bar_key_response.ok)
|
||||||
|
|
||||||
foo_data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'GET',
|
foo_data = self.client.create_temp_url(
|
||||||
container_name,
|
'GET',
|
||||||
self.object_name,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.object_name,
|
||||||
foo_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
foo_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
foo_data = self.client.create_temp_url(
|
||||||
|
'GET',
|
||||||
|
container_name,
|
||||||
|
self.object_name,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
foo_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
|
@ -890,12 +1034,21 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
foo_data.keys(),
|
foo_data.keys(),
|
||||||
msg='expires was not in the created tempurl')
|
msg='expires was not in the created tempurl')
|
||||||
|
|
||||||
bar_data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'GET',
|
bar_data = self.client.create_temp_url(
|
||||||
container_name,
|
'GET',
|
||||||
self.object_name,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.object_name,
|
||||||
bar_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
bar_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
bar_data = self.client.create_temp_url(
|
||||||
|
'GET',
|
||||||
|
container_name,
|
||||||
|
self.object_name,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
bar_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
|
@ -930,8 +1083,9 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
bar_get_response.content,
|
bar_get_response.content,
|
||||||
msg='object data was changed')
|
msg='object data was changed')
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_tempurl_expiration(self):
|
def ddtest_tempurl_expiration(self, sha_type=None):
|
||||||
"""
|
"""
|
||||||
Scenario:
|
Scenario:
|
||||||
Create a 'GET' TempURL for an existing object.
|
Create a 'GET' TempURL for an existing object.
|
||||||
|
@ -947,12 +1101,21 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
self.object_name,
|
self.object_name,
|
||||||
data=self.object_data)
|
data=self.object_data)
|
||||||
|
|
||||||
tempurl_data = self.client.create_temp_url(
|
if sha_type == 'sha2':
|
||||||
'GET',
|
tempurl_data = self.client.create_temp_url(
|
||||||
container_name,
|
'GET',
|
||||||
self.object_name,
|
container_name,
|
||||||
TEMPURL_KEY_LIFE,
|
self.object_name,
|
||||||
self.tempurl_key)
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'GET',
|
||||||
|
container_name,
|
||||||
|
self.object_name,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
self.assertIn(
|
self.assertIn(
|
||||||
'target_url',
|
'target_url',
|
||||||
|
@ -971,10 +1134,6 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
'temp_url_expires': tempurl_data.get('expires')}
|
'temp_url_expires': tempurl_data.get('expires')}
|
||||||
response = self.http.get(tempurl_data.get('target_url'), params=params)
|
response = self.http.get(tempurl_data.get('target_url'), params=params)
|
||||||
|
|
||||||
expected_disposition = 'attachment; filename="{0}"'.format(
|
|
||||||
self.object_name)
|
|
||||||
received_disposition = response.headers.get('content-disposition')
|
|
||||||
|
|
||||||
self.assertTrue(response.ok, 'object should be retrieved over tempurl')
|
self.assertTrue(response.ok, 'object should be retrieved over tempurl')
|
||||||
|
|
||||||
expected_disposition = EXPECTED_DISPOSITION.format(
|
expected_disposition = EXPECTED_DISPOSITION.format(
|
||||||
|
@ -1005,8 +1164,9 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
401,
|
401,
|
||||||
msg='tempurl did not expire')
|
msg='tempurl did not expire')
|
||||||
|
|
||||||
|
@data_driven_test(sha_type)
|
||||||
@ObjectStorageFixture.required_features('tempurl')
|
@ObjectStorageFixture.required_features('tempurl')
|
||||||
def test_create_object_manifest_with_tempurl(self):
|
def ddtest_create_object_manifest_with_tempurl(self, sha_type=None):
|
||||||
"""
|
"""
|
||||||
Scenario:
|
Scenario:
|
||||||
Test a tempurl vulnerability by creating a tempurl and then
|
Test a tempurl vulnerability by creating a tempurl and then
|
||||||
|
@ -1027,11 +1187,21 @@ class TempUrlTest(ObjectStorageFixture):
|
||||||
self.client.create_object(
|
self.client.create_object(
|
||||||
container_name, self.object_name, data=self.object_data)
|
container_name, self.object_name, data=self.object_data)
|
||||||
|
|
||||||
tempurl_data = self.client.create_temp_url('PUT',
|
if sha_type == 'sha2':
|
||||||
container_name,
|
tempurl_data = self.client.create_temp_url(
|
||||||
self.object_name,
|
'PUT',
|
||||||
TEMPURL_KEY_LIFE,
|
container_name,
|
||||||
self.tempurl_key)
|
self.object_name,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key,
|
||||||
|
sha_type='sha2')
|
||||||
|
else:
|
||||||
|
tempurl_data = self.client.create_temp_url(
|
||||||
|
'PUT',
|
||||||
|
container_name,
|
||||||
|
self.object_name,
|
||||||
|
TEMPURL_KEY_LIFE,
|
||||||
|
self.tempurl_key)
|
||||||
|
|
||||||
headers = {'Content-Length': "0",
|
headers = {'Content-Length': "0",
|
||||||
'X-Object-Manifest': '{0}/e'.format(exploit_container_name)}
|
'X-Object-Manifest': '{0}/e'.format(exploit_container_name)}
|
||||||
|
|
Loading…
Reference in New Issue