# Copyright 2013 IBM Corp.
#
# 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 ipaddress
import netaddr
from oslo_log import log as logging
from tempest.lib.common import cred_client
from tempest.lib.common import cred_provider
from tempest.lib.common.utils import data_utils
from tempest.lib import exceptions as lib_exc
from tempest.lib.services import clients
LOG = logging.getLogger(__name__)
[docs]
class DynamicCredentialProvider(cred_provider.CredentialProvider):
"""Creates credentials dynamically for tests
A credential provider that, based on an initial set of
admin credentials, creates new credentials on the fly for
tests to use and then discard.
:param str identity_version: identity API version to use `v2` or `v3`
:param str admin_role: name of the admin role added to admin users
:param str name: names of dynamic resources include this parameter
when specified
:param str credentials_domain: name of the domain where the users
are created. If not defined, the project
domain from admin_credentials is used
:param dict network_resources: network resources to be created for
the created credentials
:param Credentials admin_creds: initial admin credentials
:param bool identity_admin_domain_scope: Set to true if admin should be
scoped to the domain. By
default this is False and the
admin role is scoped to the
project.
:param str identity_admin_role: The role name to use for admin
:param list extra_roles: A list of strings for extra roles that should
be assigned to all created users
:param bool neutron_available: Whether we are running in an environment
with neutron
:param bool create_networks: Whether dynamic project networks should be
created or not
:param project_network_cidr: The CIDR to use for created project
networks
:param project_network_mask_bits: The network mask bits to use for
created project networks
:param public_network_id: The id for the public network to use
:param identity_admin_endpoint_type: The endpoint type for identity
admin clients. Defaults to public.
:param identity_uri: Identity URI of the target cloud
"""
def __init__(self, identity_version, name=None, network_resources=None,
credentials_domain=None, admin_role=None, admin_creds=None,
identity_admin_domain_scope=False,
identity_admin_role='admin', extra_roles=None,
neutron_available=False, create_networks=True,
project_network_cidr=None, project_network_mask_bits=None,
public_network_id=None, resource_prefix=None,
identity_admin_endpoint_type='public', identity_uri=None):
super(DynamicCredentialProvider, self).__init__(
identity_version=identity_version, identity_uri=identity_uri,
admin_role=admin_role, name=name,
credentials_domain=credentials_domain,
network_resources=network_resources)
self.network_resources = network_resources
self._creds = {}
self.ports = []
self.resource_prefix = resource_prefix or ''
self.neutron_available = neutron_available
self.create_networks = create_networks
self.project_network_cidr = project_network_cidr
self.project_network_mask_bits = project_network_mask_bits
self.public_network_id = public_network_id
self.default_admin_creds = admin_creds
self.identity_admin_domain_scope = identity_admin_domain_scope
self.identity_admin_role = identity_admin_role or 'admin'
self.identity_admin_endpoint_type = identity_admin_endpoint_type
self.extra_roles = extra_roles or []
(self.identity_admin_client,
self.tenants_admin_client,
self.users_admin_client,
self.roles_admin_client,
self.domains_admin_client,
self.networks_admin_client,
self.routers_admin_client,
self.subnets_admin_client,
self.ports_admin_client,
self.security_groups_admin_client) = self._get_admin_clients(
identity_admin_endpoint_type)
# Domain where isolated credentials are provisioned (v3 only).
# Use that of the admin account is None is configured.
self.creds_domain_name = None
if self.identity_version == 'v3':
self.creds_domain_name = (
self.default_admin_creds.project_domain_name or
self.credentials_domain)
self.creds_client = cred_client.get_creds_client(
self.identity_admin_client,
self.tenants_admin_client,
self.users_admin_client,
self.roles_admin_client,
self.domains_admin_client,
self.creds_domain_name)
def _get_admin_clients(self, endpoint_type):
"""Returns a tuple with instances of the following admin clients
(in this order):
identity
network
"""
os = clients.ServiceClients(self.default_admin_creds,
self.identity_uri)
params = {'endpoint_type': endpoint_type}
if self.identity_version == 'v2':
return (os.identity_v2.IdentityClient(**params),
os.identity_v2.TenantsClient(**params),
os.identity_v2.UsersClient(**params),
os.identity_v2.RolesClient(**params), None,
os.network.NetworksClient(),
os.network.RoutersClient(),
os.network.SubnetsClient(),
os.network.PortsClient(),
os.network.SecurityGroupsClient())
else:
# We use a dedicated client manager for identity client in case we
# need a different token scope for them.
if self.default_admin_creds.system:
scope = 'system'
elif (self.identity_admin_domain_scope and
(self.default_admin_creds.domain_id or
self.default_admin_creds.domain_name)):
scope = 'domain'
else:
scope = 'project'
identity_os = clients.ServiceClients(self.default_admin_creds,
self.identity_uri,
scope=scope)
return (identity_os.identity_v3.IdentityClient(**params),
identity_os.identity_v3.ProjectsClient(**params),
identity_os.identity_v3.UsersClient(**params),
identity_os.identity_v3.RolesClient(**params),
identity_os.identity_v3.DomainsClient(**params),
os.network.NetworksClient(),
os.network.RoutersClient(),
os.network.SubnetsClient(),
os.network.PortsClient(),
os.network.SecurityGroupsClient())
def _create_creds(self, admin=False, roles=None, scope='project',
project_id=None):
"""Create credentials with random name.
Creates user and role assignments on a project, domain, or system. When
the admin flag is True, creates user with the admin role on the
resource. If roles are provided, assigns those roles on the resource.
Otherwise, assigns the user the 'member' role on the resource.
:param admin: Flag if to assign to the user admin role
:type admin: bool
:param roles: Roles to assign for the user
:type roles: list
:param str scope: The scope for the role assignment, may be one of
'project', 'domain', or 'system'.
:param str project_id: The project id of already created project
for credentials under same project.
:return: Readonly Credentials with network resources
:raises: Exception if scope is invalid
"""
if not roles:
roles = []
root = self.name
cred_params = {
'project': None,
'domain': None,
'system': None
}
if scope == 'project':
if not project_id:
project_name = data_utils.rand_name(
root, prefix=self.resource_prefix)
project_desc = project_name + '-desc'
project = self.creds_client.create_project(
name=project_name, description=project_desc)
else:
# NOTE(gmann) This is the case where creds are requested
# from the existing creds within same project. We should
# not create the new project in this case.
project = self.creds_client.show_project(project_id)
project_name = project['name']
LOG.info("Using the existing project %s for scope %s and "
"roles: %s", project['id'], scope, roles)
# NOTE(andreaf) User and project can be distinguished from the
# context, having the same ID in both makes it easier to match them
# and debug.
username = project_name + '-project'
cred_params['project'] = project
elif scope == 'domain':
domain_name = data_utils.rand_name(
root, prefix=self.resource_prefix)
domain_desc = domain_name + '-desc'
domain = self.creds_client.create_domain(
name=domain_name, description=domain_desc)
username = domain_name + '-domain'
cred_params['domain'] = domain
elif scope == 'system':
prefix = data_utils.rand_name(root, prefix=self.resource_prefix)
username = prefix + '-system'
cred_params['system'] = 'all'
else:
raise lib_exc.InvalidScopeType(scope=scope)
if admin:
username += '-admin'
elif roles and len(roles) == 1:
username += '-' + roles[0]
user_password = data_utils.rand_password()
cred_params['password'] = user_password
user = self.creds_client.create_user(
username, user_password)
cred_params['user'] = user
roles_to_assign = [r for r in roles]
if admin:
roles_to_assign.append(self.admin_role)
if scope == 'project':
self.creds_client.assign_user_role(
user, project, self.identity_admin_role)
if (self.identity_version == 'v3' and
self.identity_admin_domain_scope):
self.creds_client.assign_user_role_on_domain(
user, self.identity_admin_role)
# Add roles specified in config file
roles_to_assign.extend(self.extra_roles)
# If there are still no roles, default to 'member'
# NOTE(mtreinish) For a user to have access to a project with v3 auth
# it must beassigned a role on the project. So we need to ensure that
# our newly created user has a role on the newly created project.
if not roles_to_assign and self.identity_version == 'v3':
roles_to_assign = ['member']
try:
self.creds_client.create_user_role('member')
except lib_exc.Conflict:
LOG.warning('member role already exists, ignoring conflict.')
for role in roles_to_assign:
if scope == 'project':
self.creds_client.assign_user_role(user, project, role)
elif scope == 'domain':
self.creds_client.assign_user_role_on_domain(
user, role, domain)
elif scope == 'system':
self.creds_client.assign_user_role_on_system(user, role)
LOG.info("Dynamic test user %s is created with scope %s and roles: %s",
user['id'], scope, roles_to_assign)
creds = self.creds_client.get_credentials(**cred_params)
return cred_provider.TestResources(creds)
def _create_network_resources(self, tenant_id):
"""The function creates network resources in the given tenant.
The function checks if network_resources class member is empty,
In case it is, it will create a network, a subnet and a router for
the tenant according to the given tenant id parameter.
Otherwise it will create a network resource according
to the values from network_resources dict.
:param tenant_id: The tenant id to create resources for.
:type tenant_id: str
:raises: InvalidConfiguration, Exception
:returns: network resources(network,subnet,router)
:rtype: tuple
"""
network = None
subnet = None
router = None
# Make sure settings
if self.network_resources:
if self.network_resources['router']:
if (not self.network_resources['subnet'] or
not self.network_resources['network']):
raise lib_exc.InvalidConfiguration(
'A router requires a subnet and network')
elif self.network_resources['subnet']:
if not self.network_resources['network']:
raise lib_exc.InvalidConfiguration(
'A subnet requires a network')
elif self.network_resources['dhcp']:
raise lib_exc.InvalidConfiguration('DHCP requires a subnet')
rand_name_root = data_utils.rand_name(
self.name, prefix=self.resource_prefix)
if not self.network_resources or self.network_resources['network']:
network_name = rand_name_root + "-network"
network = self._create_network(network_name, tenant_id)
try:
if not self.network_resources or self.network_resources['subnet']:
subnet_name = rand_name_root + "-subnet"
subnet = self._create_subnet(subnet_name, tenant_id,
network['id'])
if not self.network_resources or self.network_resources['router']:
router_name = rand_name_root + "-router"
router = self._create_router(router_name, tenant_id)
self._add_router_interface(router['id'], subnet['id'])
except Exception:
try:
if router:
self._clear_isolated_router(router['id'], router['name'])
if subnet:
self._clear_isolated_subnet(subnet['id'], subnet['name'])
if network:
self._clear_isolated_network(network['id'],
network['name'])
except Exception as cleanup_exception:
msg = "There was an exception trying to setup network " \
"resources for tenant %s, and this error happened " \
"trying to clean them up: %s"
LOG.warning(msg, tenant_id, cleanup_exception)
raise
return network, subnet, router
def _create_network(self, name, tenant_id):
resp_body = self.networks_admin_client.create_network(
name=name, tenant_id=tenant_id)
return resp_body['network']
def _create_subnet(self, subnet_name, tenant_id, network_id):
base_cidr = netaddr.IPNetwork(self.project_network_cidr)
mask_bits = self.project_network_mask_bits
for subnet_cidr in base_cidr.subnet(mask_bits):
try:
if self.network_resources:
resp_body = self.subnets_admin_client.\
create_subnet(
network_id=network_id, cidr=str(subnet_cidr),
name=subnet_name,
tenant_id=tenant_id,
enable_dhcp=self.network_resources['dhcp'],
ip_version=(ipaddress.ip_network(
str(subnet_cidr)).version))
else:
resp_body = self.subnets_admin_client.\
create_subnet(network_id=network_id,
cidr=str(subnet_cidr),
name=subnet_name,
tenant_id=tenant_id,
ip_version=(ipaddress.ip_network(
str(subnet_cidr)).version))
break
except lib_exc.BadRequest as e:
if 'overlaps with another subnet' not in str(e):
raise
else:
message = 'Available CIDR for subnet creation could not be found'
raise Exception(message)
return resp_body['subnet']
def _create_router(self, router_name, tenant_id):
kwargs = {'name': router_name,
'tenant_id': tenant_id}
if self.public_network_id:
kwargs['external_gateway_info'] = dict(
network_id=self.public_network_id)
resp_body = self.routers_admin_client.create_router(**kwargs)
return resp_body['router']
def _add_router_interface(self, router_id, subnet_id):
self.routers_admin_client.add_router_interface(router_id,
subnet_id=subnet_id)
def _get_project_id(self, credential_type, scope):
same_creds = [['admin'], ['member'], ['reader']]
same_alt_creds = [['alt_admin'], ['alt_member'], ['alt_reader']]
search_in = []
if credential_type in same_creds:
search_in = same_creds
elif credential_type in same_alt_creds:
search_in = same_alt_creds
for cred in search_in:
found_cred = self._creds.get("%s_%s" % (scope, str(cred)))
if found_cred:
project_id = found_cred.get("%s_%s" % (scope, 'id'))
LOG.debug("Reusing existing project %s from creds: %s ",
project_id, found_cred)
return project_id
return None
def get_credentials(self, credential_type, scope=None, by_role=False):
cred_prefix = ''
if by_role:
cred_prefix = 'role_'
if not scope and self._creds.get(
"%s%s" % (cred_prefix, str(credential_type))):
credentials = self._creds[
"%s%s" % (cred_prefix, str(credential_type))]
elif scope and (self._creds.get(
"%s%s_%s" % (cred_prefix, scope, str(credential_type)))):
credentials = self._creds[
"%s%s_%s" % (cred_prefix, scope, str(credential_type))]
else:
LOG.debug("Creating new dynamic creds for scope: %s and "
"credential_type: %s", scope, credential_type)
project_id = None
if scope:
if scope == 'project':
project_id = self._get_project_id(
credential_type, 'project')
if by_role:
credentials = self._create_creds(
roles=credential_type, scope=scope)
elif credential_type in [['admin'], ['alt_admin']]:
credentials = self._create_creds(
admin=True, scope=scope, project_id=project_id)
elif credential_type in [['alt_member'], ['alt_reader']]:
cred_type = credential_type[0][4:]
if isinstance(cred_type, str):
cred_type = [cred_type]
credentials = self._create_creds(
roles=cred_type, scope=scope, project_id=project_id)
elif credential_type in [['member'], ['reader']]:
credentials = self._create_creds(
roles=credential_type, scope=scope,
project_id=project_id)
elif credential_type in ['primary', 'alt', 'admin']:
is_admin = (credential_type == 'admin')
credentials = self._create_creds(admin=is_admin)
else:
credentials = self._create_creds(roles=credential_type)
if scope:
self._creds["%s%s_%s" % (
cred_prefix, scope, str(credential_type))] = credentials
else:
self._creds[
"%s%s" % (cred_prefix, str(credential_type))] = credentials
# Maintained until tests are ported
LOG.info("Acquired dynamic creds:\n"
" credentials: %s", credentials)
# NOTE(gmann): For 'domain' and 'system' scoped token, there is no
# project_id so we are skipping the network creation for both
# scope.
# We need to create network resource once per project.
if (not project_id and (not scope or scope == 'project')):
if (self.neutron_available and self.create_networks):
network, subnet, router = self._create_network_resources(
credentials.tenant_id)
credentials.set_resources(network=network, subnet=subnet,
router=router)
LOG.info("Created isolated network resources for:\n"
" credentials: %s", credentials)
else:
LOG.info("Network resources are not created for requested "
"scope: %s and credentials: %s", scope, credentials)
return credentials
# TODO(gmann): Remove this method in favor of get_project_member_creds()
# after the deprecation phase.
def get_primary_creds(self):
return self.get_project_member_creds()
def get_admin_creds(self):
return self.get_credentials('admin')
# TODO(gmann): Remove this method in favor of
# get_project_alt_member_creds() after the deprecation phase.
def get_alt_creds(self):
return self.get_project_alt_member_creds()
def get_system_admin_creds(self):
return self.get_credentials(['admin'], scope='system')
def get_system_member_creds(self):
return self.get_credentials(['member'], scope='system')
def get_system_reader_creds(self):
return self.get_credentials(['reader'], scope='system')
def get_domain_admin_creds(self):
return self.get_credentials(['admin'], scope='domain')
def get_domain_member_creds(self):
return self.get_credentials(['member'], scope='domain')
def get_domain_reader_creds(self):
return self.get_credentials(['reader'], scope='domain')
def get_project_admin_creds(self):
return self.get_credentials(['admin'], scope='project')
def get_project_alt_admin_creds(self):
return self.get_credentials(['alt_admin'], scope='project')
def get_project_member_creds(self):
return self.get_credentials(['member'], scope='project')
def get_project_alt_member_creds(self):
return self.get_credentials(['alt_member'], scope='project')
def get_project_reader_creds(self):
return self.get_credentials(['reader'], scope='project')
def get_project_alt_reader_creds(self):
return self.get_credentials(['alt_reader'], scope='project')
def get_creds_by_roles(self, roles, force_new=False, scope=None):
roles = list(set(roles))
# The roles list as a str will become the index as the dict key for
# the created credentials set in the dynamic_creds dict.
creds_name = "role_%s" % str(roles)
if scope:
creds_name = "role_%s_%s" % (scope, str(roles))
exist_creds = self._creds.get(creds_name)
# If force_new flag is True 2 cred sets with the same roles are needed
# handle this by creating a separate index for old one to store it
# separately for cleanup
if exist_creds and force_new:
new_index = creds_name + '-' + str(len(self._creds))
self._creds[new_index] = exist_creds
del self._creds[creds_name]
return self.get_credentials(roles, scope=scope, by_role=True)
def _clear_isolated_router(self, router_id, router_name):
client = self.routers_admin_client
try:
client.delete_router(router_id)
except lib_exc.NotFound:
LOG.warning('router with name: %s not found for delete',
router_name)
def _clear_isolated_subnet(self, subnet_id, subnet_name):
client = self.subnets_admin_client
try:
client.delete_subnet(subnet_id)
except lib_exc.NotFound:
LOG.warning('subnet with name: %s not found for delete',
subnet_name)
def _clear_isolated_network(self, network_id, network_name):
net_client = self.networks_admin_client
try:
net_client.delete_network(network_id)
except lib_exc.NotFound:
LOG.warning('network with name: %s not found for delete',
network_name)
def _clear_isolated_net_resources(self):
client = self.routers_admin_client
for cred in self._creds:
creds = self._creds.get(cred)
if (not creds or not any([creds.router, creds.network,
creds.subnet])):
continue
LOG.debug("Clearing network: %(network)s, "
"subnet: %(subnet)s, router: %(router)s",
{'network': creds.network, 'subnet': creds.subnet,
'router': creds.router})
if (not self.network_resources or
(self.network_resources.get('router') and creds.subnet)):
try:
client.remove_router_interface(
creds.router['id'],
subnet_id=creds.subnet['id'])
except lib_exc.NotFound:
LOG.warning('router with name: %s not found for delete',
creds.router['name'])
self._clear_isolated_router(creds.router['id'],
creds.router['name'])
if (not self.network_resources or
self.network_resources.get('subnet')):
self._clear_isolated_subnet(creds.subnet['id'],
creds.subnet['name'])
if (not self.network_resources or
self.network_resources.get('network')):
self._clear_isolated_network(creds.network['id'],
creds.network['name'])
def clear_creds(self):
if not self._creds:
return
self._clear_isolated_net_resources()
project_ids = set()
for creds in self._creds.values():
# NOTE(gmann): With new RBAC personas, we can have single project
# and multiple user created under it, to avoid conflict let's
# cleanup the projects at the end.
# Adding project if id is not None, means leaving domain and
# system creds.
if creds.project_id:
project_ids.add(creds.project_id)
try:
self.creds_client.delete_user(creds.user_id)
except lib_exc.NotFound:
LOG.warning("user with name: %s not found for delete",
creds.username)
# if cred is domain scoped, delete ephemeral domain
# do not delete default domain
if (hasattr(creds, 'domain_id') and
creds.domain_id != creds.project_domain_id):
try:
self.creds_client.delete_domain(creds.domain_id)
except lib_exc.NotFound:
LOG.warning("domain with name: %s not found for delete",
creds.domain_name)
for project_id in project_ids:
# NOTE(zhufl): Only when neutron's security_group ext is
# enabled, cleanup_default_secgroup will not raise error. But
# here cannot use test_utils.is_extension_enabled for it will
# cause "circular dependency". So here just use try...except to
# ensure tenant deletion without big changes.
LOG.info("Deleting project and security group for project: %s",
project_id)
try:
if self.neutron_available:
self.cleanup_default_secgroup(
self.security_groups_admin_client, project_id)
except lib_exc.NotFound:
LOG.warning("failed to cleanup tenant %s's secgroup",
project_id)
try:
self.creds_client.delete_project(project_id)
except lib_exc.NotFound:
LOG.warning("tenant with id: %s not found for delete",
project_id)
self._creds = {}
def is_multi_user(self):
return True
def is_multi_tenant(self):
return True
def is_role_available(self, role):
return True
