openstack
/
designate
Code Issues Proposed changes

Merge "Unauthed tests for RecordSets (Security)"

This commit is contained in:
Jenkins 2015-09-22 15:23:01 +00:00 committed by Gerrit Code Review
parent fb0dfdbb7e 1e6456ac1f
commit 0ecb86e437
2 changed files with 127 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

90
functionaltests/api/v2/test_recordset_unauthed.py Normal file
View File

@ -0,0 +1,90 @@
"""
Copyright 2015 Rackspace
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.
"""
from tempest_lib import exceptions
from functionaltests.common import datagen
from functionaltests.common import utils
from functionaltests.api.v2.base import DesignateV2Test
from functionaltests.api.v2.clients.recordset_client import RecordsetClient
from functionaltests.api.v2.clients.zone_client import ZoneClient
@utils.parameterized_class
class RecordsetTest(DesignateV2Test):
def setUp(self):
super(RecordsetTest, self).setUp()
self.increase_quotas(user='default')
resp, self.zone = ZoneClient.as_user('default').post_zone(
datagen.random_zone_data())
ZoneClient.as_user('default').wait_for_zone(self.zone.id)
self.client = RecordsetClient.as_user('default', with_token=False)
def tearDown(self):
super(RecordsetTest, self).tearDown()
resp, self.zone = ZoneClient.as_user('default').delete_zone(
self.zone.id)
def test_create_a_recordset(self):
post_model = datagen.random_a_recordset(self.zone.name)
self.assertRaises(
exceptions.Unauthorized, self.client.post_recordset, self.zone.id,
post_model)
def test_get_fake_recordset(self):
self.assertRaises(
exceptions.Unauthorized, self.client.get_recordset, self.zone.id,
'junk')
def test_get_existing_recordset(self):
post_model = datagen.random_a_recordset(self.zone.name)
resp, resp_model = RecordsetClient.as_user('default') \
.post_recordset(self.zone.id, post_model)
self.assertRaises(
exceptions.Unauthorized, self.client.get_recordset, self.zone.id,
resp_model.id)
def test_list_recordsets(self):
self.assertRaises(
exceptions.Unauthorized, self.client.list_recordsets, self.zone.id)
def test_update_fake_recordset(self):
put_model = datagen.random_a_recordset(self.zone.name)
self.assertRaises(
exceptions.Unauthorized, self.client.put_recordset, self.zone.id,
'junk', put_model)
def test_update_existing_recordset(self):
post_model = datagen.random_a_recordset(self.zone.name)
resp, resp_model = RecordsetClient.as_user('default') \
.post_recordset(self.zone.id, post_model)
self.assertRaises(
exceptions.Unauthorized, self.client.put_recordset, self.zone.id,
resp_model.id, post_model)
def test_delete_fake_recordset(self):
self.assertRaises(
exceptions.Unauthorized, self.client.delete_recordset,
self.zone.id, 'junk')
def test_delete_existing_recordset(self):
post_model = datagen.random_a_recordset(self.zone.name)
resp, resp_model = RecordsetClient.as_user('default') \
.post_recordset(self.zone.id, post_model)
self.assertRaises(
exceptions.Unauthorized, self.client.delete_recordset,
self.zone.id, resp_model.id)

56
functionaltests/common/client.py
View File

@ -38,19 +38,31 @@ class KeystoneV2AuthProviderWithOverridableUrl(KeystoneV2AuthProvider):
.base_url(*args, **kwargs)
class KeystoneV2AuthProviderNoToken(KeystoneV2AuthProviderWithOverridableUrl):
def _decorate_request(self, filters, method, url, headers=None, body=None,
auth_data=None):
_res = super(KeystoneV2AuthProviderNoToken, self)._decorate_request(
filters, method, url, headers=headers, body=body,
auth_data=auth_data)
_url, _headers, _body = _res
del _headers['X-Auth-Token']
return (_url, _headers, _body)
class BaseDesignateClient(RestClient):
def __init__(self):
def __init__(self, with_token=True):
super(BaseDesignateClient, self).__init__(
auth_provider=self.get_auth_provider(),
auth_provider=self.get_auth_provider(with_token),
service='dns',
region=cfg.CONF.identity.region
)
def get_auth_provider(self):
def get_auth_provider(self, with_token=True):
if cfg.CONF.noauth.use_noauth:
return self._get_noauth_auth_provider()
return self._get_keystone_auth_provider()
return self._get_keystone_auth_provider(with_token)
@abc.abstractmethod
def _get_noauth_auth_provider(self):
@ -60,9 +72,13 @@ class BaseDesignateClient(RestClient):
def _get_keystone_auth_provider(self):
pass
def _create_keystone_auth_provider(self, creds):
auth_provider = KeystoneV2AuthProviderWithOverridableUrl(
creds, cfg.CONF.identity.uri)
def _create_keystone_auth_provider(self, creds, with_token=True):
if with_token:
auth_provider = KeystoneV2AuthProviderWithOverridableUrl(
creds, cfg.CONF.identity.uri)
else:
auth_provider = KeystoneV2AuthProviderNoToken(
creds, cfg.CONF.identity.uri)
auth_provider.fill_credentials()
return auth_provider
@ -76,13 +92,13 @@ class DesignateClient(BaseDesignateClient):
)
return NoAuthAuthProvider(creds, cfg.CONF.noauth.designate_endpoint)
def _get_keystone_auth_provider(self):
def _get_keystone_auth_provider(self, with_token=True):
creds = KeystoneV2Credentials(
username=cfg.CONF.identity.username,
password=cfg.CONF.identity.password,
tenant_name=cfg.CONF.identity.tenant_name,
)
return self._create_keystone_auth_provider(creds)
return self._create_keystone_auth_provider(creds, with_token)
class DesignateAltClient(BaseDesignateClient):
@ -94,13 +110,13 @@ class DesignateAltClient(BaseDesignateClient):
)
return NoAuthAuthProvider(creds, cfg.CONF.noauth.designate_endpoint)
def _get_keystone_auth_provider(self):
def _get_keystone_auth_provider(self, with_token=True):
creds = KeystoneV2Credentials(
username=cfg.CONF.identity.alt_username,
password=cfg.CONF.identity.alt_password,
tenant_name=cfg.CONF.identity.alt_tenant_name,
)
return self._create_keystone_auth_provider(creds)
return self._create_keystone_auth_provider(creds, with_token)
class DesignateAdminClient(BaseDesignateClient):
@ -112,24 +128,24 @@ class DesignateAdminClient(BaseDesignateClient):
)
return NoAuthAuthProvider(creds, cfg.CONF.noauth.designate_endpoint)
def _get_keystone_auth_provider(self):
def _get_keystone_auth_provider(self, with_token=True):
creds = KeystoneV2Credentials(
username=cfg.CONF.identity.admin_username,
password=cfg.CONF.identity.admin_password,
tenant_name=cfg.CONF.identity.admin_tenant_name,
)
return self._create_keystone_auth_provider(creds)
return self._create_keystone_auth_provider(creds, with_token)
class ClientMixin(object):
@classmethod
@memoized
def get_clients(cls):
def get_clients(cls, with_token):
return {
'default': DesignateClient(),
'alt': DesignateAltClient(),
'admin': DesignateAdminClient(),
'default': DesignateClient(with_token),
'alt': DesignateAltClient(with_token),
'admin': DesignateAdminClient(with_token),
}
def __init__(self, client):
@ -140,11 +156,13 @@ class ClientMixin(object):
return resp, model_type.from_json(body)
@classmethod
def as_user(cls, user):
def as_user(cls, user, with_token=True):
"""
:param user: 'default', 'alt', or 'admin'
:param with_token: Boolean for whether to send the x-auth-token with
requests
"""
return cls(cls.get_clients()[user])
return cls(cls.get_clients(with_token)[user])
@property
def tenant_id(self):