9dbeefb55e
The repo is Python 3 now, so update hacking to version 3.0 which supports Python 3. Fix problems found by updated hacking version. Update local hacking checks for new flake8, remove test B314 since that tests difference between Python 2 and 3, there's no need to advise using six anymore. Use oslotest.base directly, this fixes the hacking tests. Remove ddt usage in testsuite, it does not work with current hacking version anymore. Change-Id: Iee4584c6fde08728c017468d9de1db73f2c79d8d
568 lines
20 KiB
Python
568 lines
20 KiB
Python
# Copyright (c) 2014 Rackspace, Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
|
# implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
import copy
|
|
|
|
from testtools import testcase
|
|
|
|
from barbican.tests import utils
|
|
from functionaltests.api import base
|
|
from functionaltests.api.v1.behaviors import container_behaviors
|
|
from functionaltests.api.v1.behaviors import secret_behaviors
|
|
from functionaltests.api.v1.models import container_models
|
|
from functionaltests.api.v1.models import secret_models
|
|
|
|
|
|
def get_default_container_create_data(secret):
|
|
return {
|
|
"type": "generic",
|
|
"name": "generic name",
|
|
"secret_refs": [
|
|
{
|
|
"name": "a secret",
|
|
"secret_ref": secret
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
create_container_data = {
|
|
"name": "containername",
|
|
"type": "generic",
|
|
"secret_refs": [
|
|
{
|
|
"name": "secret1",
|
|
},
|
|
{
|
|
"name": "secret2",
|
|
},
|
|
{
|
|
"name": "secret3"
|
|
}
|
|
]
|
|
}
|
|
|
|
create_container_rsa_data = {
|
|
"name": "rsacontainer",
|
|
"type": "rsa",
|
|
"secret_refs": [
|
|
{
|
|
"name": "public_key",
|
|
},
|
|
{
|
|
"name": "private_key",
|
|
},
|
|
{
|
|
"name": "private_key_passphrase"
|
|
}
|
|
]
|
|
}
|
|
|
|
accepted_str_values = {
|
|
'alphanumeric': ['a2j3j6ll9'],
|
|
'punctuation': ['~!@#$%^&*()_+`-={}[]|:;<>,.?'],
|
|
'len_255': [base.TestCase.max_sized_field],
|
|
'uuid': ['54262d9d-4bc7-4821-8df0-dc2ca8e112bb'],
|
|
'empty': ['']
|
|
}
|
|
|
|
|
|
class BaseContainerTestCase(base.TestCase):
|
|
default_data_template = create_container_data
|
|
|
|
def setUp(self):
|
|
super(BaseContainerTestCase, self).setUp()
|
|
self.secret_behaviors = secret_behaviors.SecretBehaviors(self.client)
|
|
self.behaviors = container_behaviors.ContainerBehaviors(self.client)
|
|
|
|
# Setting up three secrets for building containers
|
|
self.secret_ref_1 = self._create_a_secret()
|
|
self.secret_ref_2 = self._create_a_secret()
|
|
self.secret_ref_3 = self._create_a_secret()
|
|
|
|
self.default_data = copy.deepcopy(self.default_data_template)
|
|
|
|
default_secret_refs = self.default_data['secret_refs']
|
|
default_secret_refs[0]['secret_ref'] = self.secret_ref_1
|
|
default_secret_refs[1]['secret_ref'] = self.secret_ref_2
|
|
default_secret_refs[2]['secret_ref'] = self.secret_ref_3
|
|
|
|
def tearDown(self):
|
|
self.secret_behaviors.delete_all_created_secrets()
|
|
self.behaviors.delete_all_created_containers()
|
|
super(BaseContainerTestCase, self).tearDown()
|
|
|
|
def _cleanup_all_containers(self):
|
|
for user_name in self.client.get_all_functional_test_user_names():
|
|
self.behaviors.delete_all_containers_for_user(user_name=user_name)
|
|
|
|
def _create_a_secret(self):
|
|
secret_defaults_data = {
|
|
"name": "AES key",
|
|
"expiration": "2030-02-28T19:14:44.180394",
|
|
"algorithm": "aes",
|
|
"bit_length": 256,
|
|
"mode": "cbc",
|
|
"payload": "gF6+lLoF3ohA9aPRpt+6bQ==",
|
|
"payload_content_type": "application/octet-stream",
|
|
"payload_content_encoding": "base64",
|
|
}
|
|
|
|
secret_model = secret_models.SecretModel(**secret_defaults_data)
|
|
resp, secret_ref = self.secret_behaviors.create_secret(secret_model)
|
|
self.assertEqual(201, resp.status_code)
|
|
self.assertIsNotNone(secret_ref)
|
|
|
|
return secret_ref
|
|
|
|
|
|
@utils.parameterized_test_case
|
|
class GenericContainersTestCase(BaseContainerTestCase):
|
|
|
|
@testcase.attr('positive')
|
|
def test_create_defaults_none_secret_name(self):
|
|
"""Covers creating a container with None as a secret name."""
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
test_model.name = None
|
|
|
|
resp, container_ref = self.behaviors.create_container(test_model)
|
|
self.assertEqual(201, resp.status_code)
|
|
|
|
@utils.parameterized_dataset({'0': [0], '1': [1], '50': [50]})
|
|
@testcase.attr('positive')
|
|
def test_create_defaults_size(self, num_secrets):
|
|
"""Covers creating containers of various sizes."""
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
for i in range(0, num_secrets):
|
|
secret_ref = self._create_a_secret()
|
|
test_model.secret_refs.append({
|
|
'name': 'other_secret{0}'.format(i),
|
|
'secret_ref': secret_ref
|
|
})
|
|
|
|
resp, container_ref = self.behaviors.create_container(test_model)
|
|
self.assertEqual(201, resp.status_code)
|
|
|
|
@utils.parameterized_dataset(accepted_str_values)
|
|
@testcase.attr('positive')
|
|
def test_create_defaults_name(self, name):
|
|
"""Covers creating generic containers with various names."""
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
test_model.name = name
|
|
|
|
resp, container_ref = self.behaviors.create_container(test_model)
|
|
self.assertEqual(201, resp.status_code)
|
|
|
|
@utils.parameterized_dataset(accepted_str_values)
|
|
@testcase.attr('positive')
|
|
def test_create_defaults_secret_name(self, name=None):
|
|
"""Covers creating containers with various secret ref names."""
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
test_model.secret_refs = [{
|
|
'name': name,
|
|
'secret_ref': self.secret_ref_1
|
|
}]
|
|
|
|
resp, container_ref = self.behaviors.create_container(test_model)
|
|
self.assertEqual(201, resp.status_code)
|
|
|
|
get_resp = self.behaviors.get_container(container_ref)
|
|
self.assertEqual(200, get_resp.status_code)
|
|
self.assertEqual(get_resp.model.secret_refs[0].name, name)
|
|
|
|
@testcase.attr('negative')
|
|
def test_create_defaults_invalid_type(self):
|
|
"""Container creating should fail with an invalid container type."""
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
test_model.type = 'bad_type'
|
|
|
|
resp, container_ref = self.behaviors.create_container(test_model)
|
|
self.assertEqual(400, resp.status_code)
|
|
|
|
@testcase.attr('negative')
|
|
def test_create_defaults_duplicate_secret_refs(self):
|
|
"""Covers creating a container with a duplicated secret ref."""
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
test_model.secret_refs[0]['secret_ref'] = self.secret_ref_1
|
|
test_model.secret_refs[1]['secret_ref'] = self.secret_ref_1
|
|
test_model.secret_refs[2]['secret_ref'] = self.secret_ref_1
|
|
|
|
resp, container_ref = self.behaviors.create_container(test_model)
|
|
self.assertEqual(400, resp.status_code)
|
|
|
|
@testcase.attr('negative')
|
|
def test_get_non_existent_container(self):
|
|
"""A get on a container that does not exist should return a 404."""
|
|
resp = self.behaviors.get_container("not_a_ref")
|
|
self.assertEqual(404, resp.status_code)
|
|
|
|
@testcase.attr('negative')
|
|
def test_delete_non_existent_container(self):
|
|
"""A delete on a container that does not exist should return a 404."""
|
|
resp = self.behaviors.delete_container("not_a_ref", expected_fail=True)
|
|
self.assertEqual(404, resp.status_code)
|
|
|
|
@testcase.skipIf(not base.conf_host_href_used, 'response href using '
|
|
'wsgi request instead of CONF.host_href')
|
|
@testcase.attr('positive')
|
|
def test_create_change_host_with_header_not_allowed(self, **kwargs):
|
|
"""Create a container with a (possibly) malicious host name header."""
|
|
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
|
|
malicious_hostname = 'some.bad.server.com'
|
|
changed_host_header = {'Host': malicious_hostname}
|
|
|
|
resp, container_ref = self.behaviors.create_container(
|
|
test_model, extra_headers=changed_host_header)
|
|
|
|
self.assertEqual(201, resp.status_code)
|
|
|
|
# get Location field from result and assert that it is NOT the
|
|
# malicious one.
|
|
regex = '.*{0}.*'.format(malicious_hostname)
|
|
self.assertNotRegex(resp.headers['location'], regex)
|
|
|
|
@testcase.skipIf(base.conf_host_href_used, 'response href using '
|
|
'CONF.host_href instead of wsgi request')
|
|
@testcase.attr('positive')
|
|
def test_get_change_host_with_header_allowed(self, **kwargs):
|
|
"""Get a container with a alternative proxy host name header."""
|
|
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
|
|
another_proxy_hostname = 'proxy2.server.com'
|
|
changed_host_header = {'Host': another_proxy_hostname}
|
|
|
|
# In test, cannot pass different host header during create as returned
|
|
# container_href in response contains that host in url. That url is
|
|
# used in deleting that container during cleanup step.
|
|
resp, container_href = self.behaviors.create_container(test_model)
|
|
self.assertEqual(201, resp.status_code)
|
|
|
|
resp = self.behaviors.get_container(container_href,
|
|
extra_headers=changed_host_header)
|
|
# Assert that returned href has provided proxy hostname
|
|
regex = '.*{0}.*'.format(another_proxy_hostname)
|
|
self.assertRegex(resp.model.container_ref, regex)
|
|
|
|
|
|
@utils.parameterized_test_case
|
|
class RSAContainersTestCase(BaseContainerTestCase):
|
|
default_data_template = create_container_rsa_data
|
|
|
|
@testcase.attr('positive')
|
|
def test_create_rsa_no_passphrase(self):
|
|
"""Covers creating an rsa container without a passphrase."""
|
|
pub_key_ref = {'name': 'public_key', 'secret_ref': self.secret_ref_1}
|
|
priv_key_ref = {'name': 'private_key', 'secret_ref': self.secret_ref_2}
|
|
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
test_model.secret_refs = [pub_key_ref, priv_key_ref]
|
|
|
|
resp, container_ref = self.behaviors.create_container(test_model)
|
|
self.assertEqual(201, resp.status_code)
|
|
|
|
@utils.parameterized_dataset(accepted_str_values)
|
|
@testcase.attr('positive')
|
|
def test_create_rsa_name(self, name):
|
|
"""Covers creating rsa containers with various names."""
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
test_model.name = name
|
|
|
|
resp, container_ref = self.behaviors.create_container(test_model)
|
|
self.assertEqual(201, resp.status_code)
|
|
|
|
@testcase.attr('negative')
|
|
def test_create_rsa_invalid_key_names(self):
|
|
"""Covers creating an RSA container with incorrect names."""
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
test_model.secret_refs = [
|
|
{
|
|
"name": "secret1",
|
|
"secret_ref": self.secret_ref_1
|
|
},
|
|
{
|
|
"name": "secret2",
|
|
"secret_ref": self.secret_ref_2
|
|
},
|
|
{
|
|
"name": "secret3",
|
|
"secret_ref": self.secret_ref_3
|
|
}
|
|
]
|
|
|
|
resp, container_ref = self.behaviors.create_container(test_model)
|
|
self.assertEqual(400, resp.status_code)
|
|
|
|
@testcase.attr('negative')
|
|
def test_create_rsa_no_public_key(self):
|
|
"""Creating an rsa container without a public key should fail.
|
|
|
|
RSA containers must have at least a public key and private key.
|
|
"""
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
test_model.secret_refs[0]['name'] = 'secret_1'
|
|
|
|
resp, container_ref = self.behaviors.create_container(test_model)
|
|
self.assertEqual(400, resp.status_code)
|
|
|
|
@testcase.attr('negative')
|
|
def test_create_rsa_no_private_key(self):
|
|
"""Creating an rsa container without a private key should fail.
|
|
|
|
RSA containers must have at least a public key and private key.
|
|
"""
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
test_model.secret_refs[1]['name'] = 'secret_1'
|
|
|
|
resp, container_ref = self.behaviors.create_container(test_model)
|
|
self.assertEqual(400, resp.status_code)
|
|
|
|
@testcase.skipIf(not base.conf_host_href_used, 'response href using '
|
|
'wsgi request instead of CONF.host_href')
|
|
@testcase.attr('positive')
|
|
def test_create_rsa_change_host_with_header_not_allowed(self, **kwargs):
|
|
"""Create a container with a (possibly) malicious host name header."""
|
|
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
|
|
malicious_hostname = 'some.bad.server.com'
|
|
changed_host_header = {'Host': malicious_hostname}
|
|
|
|
resp, container_ref = self.behaviors.create_container(
|
|
test_model, extra_headers=changed_host_header)
|
|
|
|
self.assertEqual(201, resp.status_code)
|
|
|
|
# get Location field from result and assert that it is NOT the
|
|
# malicious one.
|
|
regex = '.*{0}.*'.format(malicious_hostname)
|
|
self.assertNotRegex(resp.headers['location'], regex)
|
|
|
|
@testcase.skipIf(base.conf_host_href_used, 'response href using '
|
|
'CONF.host_href instead of wsgi request')
|
|
@testcase.attr('positive')
|
|
def test_get_rsa_change_host_with_header_allowed(self, **kwargs):
|
|
"""Get a container with a alternative proxy host name header."""
|
|
|
|
test_model = container_models.ContainerModel(**self.default_data)
|
|
|
|
another_proxy_hostname = 'proxy2.server.com'
|
|
changed_host_header = {'Host': another_proxy_hostname}
|
|
|
|
# In test, cannot pass different host header during create as returned
|
|
# container_href in response contains that host in url. That url is
|
|
# used in deleting that container during cleanup step.
|
|
resp, container_href = self.behaviors.create_container(test_model)
|
|
self.assertEqual(201, resp.status_code)
|
|
|
|
resp = self.behaviors.get_container(container_href,
|
|
extra_headers=changed_host_header)
|
|
# Assert that returned href has provided proxy hostname
|
|
regex = '.*{0}.*'.format(another_proxy_hostname)
|
|
self.assertRegex(resp.model.container_ref, regex)
|
|
|
|
|
|
class ContainersPagingTestCase(base.PagingTestCase):
|
|
|
|
def setUp(self):
|
|
super(ContainersPagingTestCase, self).setUp()
|
|
self.secret_behaviors = secret_behaviors.SecretBehaviors(self.client)
|
|
self.behaviors = container_behaviors.ContainerBehaviors(self.client)
|
|
|
|
# make a local mutable copy of the default data to prevent
|
|
# possible data contamination
|
|
secret = self._create_a_secret()
|
|
self.create_default_data = get_default_container_create_data(secret)
|
|
|
|
def _create_a_secret(self):
|
|
secret_defaults_data = {
|
|
"name": "AES key",
|
|
"expiration": "2030-02-28T19:14:44.180394",
|
|
"algorithm": "aes",
|
|
"bit_length": 256,
|
|
"mode": "cbc",
|
|
"payload": "gF6+lLoF3ohA9aPRpt+6bQ==",
|
|
"payload_content_type": "application/octet-stream",
|
|
"payload_content_encoding": "base64",
|
|
}
|
|
|
|
secret_model = secret_models.SecretModel(**secret_defaults_data)
|
|
resp, secret_ref = self.secret_behaviors.create_secret(secret_model)
|
|
self.assertEqual(201, resp.status_code)
|
|
self.assertIsNotNone(secret_ref)
|
|
|
|
return secret_ref
|
|
|
|
def tearDown(self):
|
|
self.secret_behaviors.delete_all_created_secrets()
|
|
self.behaviors.delete_all_created_containers()
|
|
super(ContainersPagingTestCase, self).tearDown()
|
|
|
|
def create_model(self):
|
|
return container_models.ContainerModel(**self.create_default_data)
|
|
|
|
def create_resources(self, count=0, model=None):
|
|
for x in range(0, count):
|
|
self.behaviors.create_container(model)
|
|
|
|
def get_resources(self, limit=10, offset=0, filter=filter):
|
|
return self.behaviors.get_containers(limit=limit, offset=offset,
|
|
filter=filter)
|
|
|
|
def set_filter_field(self, unique_str, model):
|
|
'''Set the name field which we use in the get_resources '''
|
|
model.name = unique_str
|
|
|
|
|
|
class ContainersUnauthedTestCase(BaseContainerTestCase):
|
|
|
|
def setUp(self):
|
|
super(ContainersUnauthedTestCase, self).setUp()
|
|
|
|
self.dummy_project_id = 'dummy123'
|
|
self.dummy_container_ref = (
|
|
'containers/dummy123-3416-4b53-8875-e6af3e0af8c3'
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.secret_behaviors.delete_all_created_secrets()
|
|
super(ContainersUnauthedTestCase, self).tearDown()
|
|
|
|
@testcase.attr('negative', 'security')
|
|
def test_unauthed_create_huge_dummy_token_no_proj_id(self):
|
|
"""Attempt to create a container with a dummy token, and no project id
|
|
|
|
Should return 401
|
|
"""
|
|
|
|
model = container_models.ContainerModel(
|
|
**create_container_data
|
|
)
|
|
headers = {'X-Auth-Token': 'a' * 3500}
|
|
resp = self.client.post(
|
|
'containers', request_model=model, use_auth=False,
|
|
extra_headers=headers
|
|
)
|
|
self.assertEqual(401, resp.status_code)
|
|
|
|
@testcase.attr('negative', 'security')
|
|
def test_unauthed_create_no_proj_id(self):
|
|
"""Attempt to create a container without a token or project id
|
|
|
|
Should return 401
|
|
"""
|
|
|
|
model = container_models.ContainerModel(
|
|
**create_container_data
|
|
)
|
|
resp = self.client.post(
|
|
'containers', request_model=model, use_auth=False
|
|
)
|
|
self.assertEqual(401, resp.status_code)
|
|
|
|
@testcase.attr('negative', 'security')
|
|
def test_unauthed_get_no_proj_id(self):
|
|
"""Attempt to get a container without a token or project id
|
|
|
|
Should return 401
|
|
"""
|
|
|
|
headers = {
|
|
'Accept': '*/*',
|
|
'Accept-Encoding': '*/*'
|
|
}
|
|
resp = self.client.get(
|
|
self.dummy_container_ref, extra_headers=headers, use_auth=False
|
|
)
|
|
self.assertEqual(401, resp.status_code)
|
|
|
|
@testcase.attr('negative', 'security')
|
|
def test_unauthed_delete_no_proj_id(self):
|
|
"""Attempt to delete a container without a token or project id
|
|
|
|
Should return 401
|
|
"""
|
|
|
|
resp = self.client.delete(self.dummy_container_ref, use_auth=False)
|
|
self.assertEqual(401, resp.status_code)
|
|
|
|
@testcase.attr('negative', 'security')
|
|
def test_unauthed_huge_dummy_token_with_proj_id(self):
|
|
"""Attempt to create a container with a dummy token and project id
|
|
|
|
Should return 401
|
|
"""
|
|
|
|
model = container_models.ContainerModel(
|
|
**create_container_data
|
|
)
|
|
headers = {
|
|
'X-Auth-Token': 'a' * 3500,
|
|
'X-Project-Id': self.dummy_project_id
|
|
}
|
|
resp = self.client.post(
|
|
'containers', request_model=model, use_auth=False,
|
|
extra_headers=headers
|
|
)
|
|
self.assertEqual(401, resp.status_code)
|
|
|
|
@testcase.attr('negative', 'security')
|
|
def test_unauthed_create_with_proj_id(self):
|
|
"""Attempt to create a container with a project id, but not a token
|
|
|
|
Should return 401
|
|
"""
|
|
|
|
model = container_models.ContainerModel(
|
|
**create_container_data
|
|
)
|
|
headers = {'X-Project-Id': self.dummy_project_id}
|
|
resp = self.client.post(
|
|
'containers', request_model=model, use_auth=False,
|
|
extra_headers=headers
|
|
)
|
|
self.assertEqual(401, resp.status_code)
|
|
|
|
@testcase.attr('negative', 'security')
|
|
def test_unauthed_get_with_proj_id(self):
|
|
"""Attempt to get a container with a project id, but not a token
|
|
|
|
Should return 401
|
|
"""
|
|
|
|
headers = {
|
|
'Accept': '*/*',
|
|
'Accept-Encoding': '*/*',
|
|
'X-Project-Id': self.dummy_project_id
|
|
}
|
|
resp = self.client.get(
|
|
self.dummy_container_ref, extra_headers=headers, use_auth=False
|
|
)
|
|
self.assertEqual(401, resp.status_code)
|
|
|
|
@testcase.attr('negative', 'security')
|
|
def test_unauthed_delete_with_proj_id(self):
|
|
"""Attempt to delete a container with a project id, but not a token
|
|
|
|
Should return 401
|
|
"""
|
|
|
|
headers = {'X-Project-Id': self.dummy_project_id}
|
|
resp = self.client.delete(
|
|
self.dummy_container_ref, use_auth=False, extra_headers=headers
|
|
)
|
|
self.assertEqual(401, resp.status_code)
|