tempest/common/dynamic_creds.py

# 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 netaddr
from oslo_log import log as logging
import six
from tempest_lib import exceptions as lib_exc

from tempest import clients
from tempest.common import cred_client
from tempest.common import cred_provider
from tempest.common.utils import data_utils
from tempest import config
from tempest import exceptions

CONF = config.CONF
LOG = logging.getLogger(__name__)


class DynamicCredentialProvider(cred_provider.CredentialProvider):

    def __init__(self, identity_version, name=None, network_resources=None,
                 credentials_domain=None, admin_role=None, admin_creds=None):
        """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
        """
        super(DynamicCredentialProvider, self).__init__(
            identity_version=identity_version, admin_role=admin_role,
            name=name, credentials_domain=credentials_domain,
            network_resources=network_resources)
        self.network_resources = network_resources
        self._creds = {}
        self.ports = []
        self.default_admin_creds = admin_creds
        (self.identity_admin_client, self.tenants_admin_client,
         self.roles_admin_client,
         self.users_admin_client,
         self.network_admin_client,
         self.networks_admin_client,
         self.subnets_admin_client,
         self.ports_admin_client,
         self.security_groups_admin_client) = self._get_admin_clients()
        # 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.roles_admin_client,
            self.users_admin_client,
            self.creds_domain_name)

    def _get_admin_clients(self):
        """Returns a tuple with instances of the following admin clients

        (in this order):
            identity
            network
        """
        os = clients.Manager(self.default_admin_creds)
        if self.identity_version == 'v2':
            return (os.identity_client, os.tenants_client, os.roles_client,
                    os.users_client, os.network_client, os.networks_client,
                    os.subnets_client, os.ports_client,
                    os.security_groups_client)
        else:
            return (os.identity_v3_client, None, None, None, os.network_client,
                    os.networks_client, os.subnets_client, os.ports_client,
                    os.security_groups_client)

    def _create_creds(self, suffix="", admin=False, roles=None):
        """Create random credentials under the following schema.

        If the name contains a '.' is the full class path of something, and
        we don't really care. If it isn't, it's probably a meaningful name,
        so use it.

        For logging purposes, -user and -tenant are long and redundant,
        don't use them. The user# will be sufficient to figure it out.
        """
        if '.' in self.name:
            root = ""
        else:
            root = self.name

        project_name = data_utils.rand_name(root) + suffix
        project_desc = project_name + "-desc"
        project = self.creds_client.create_project(
            name=project_name, description=project_desc)

        username = data_utils.rand_name(root) + suffix
        user_password = data_utils.rand_password()
        email = data_utils.rand_name(root) + suffix + "@example.com"
        user = self.creds_client.create_user(
            username, user_password, project, email)
        if 'user' in user:
            user = user['user']
        role_assigned = False
        if admin:
            self.creds_client.assign_user_role(user, project,
                                               self.admin_role)
            role_assigned = True
        # Add roles specified in config file
        for conf_role in CONF.auth.tempest_roles:
            self.creds_client.assign_user_role(user, project, conf_role)
            role_assigned = True
        # Add roles requested by caller
        if roles:
            for role in roles:
                self.creds_client.assign_user_role(user, project, role)
                role_assigned = True
        # 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 self.identity_version == 'v3' and not role_assigned:
            self.creds_client.create_user_role('Member')
            self.creds_client.assign_user_role(user, project, 'Member')

        creds = self.creds_client.get_credentials(user, project, user_password)
        return cred_provider.TestResources(creds)

    def _create_network_resources(self, tenant_id):
        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 exceptions.InvalidConfiguration(
                        'A router requires a subnet and network')
            elif self.network_resources['subnet']:
                if not self.network_resources['network']:
                    raise exceptions.InvalidConfiguration(
                        'A subnet requires a network')
            elif self.network_resources['dhcp']:
                raise exceptions.InvalidConfiguration('DHCP requires a subnet')

        data_utils.rand_name_root = data_utils.rand_name(self.name)
        if not self.network_resources or self.network_resources['network']:
            network_name = data_utils.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 = data_utils.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 = data_utils.rand_name_root + "-router"
                router = self._create_router(router_name, tenant_id)
                self._add_router_interface(router['id'], subnet['id'])
        except Exception:
            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'])
            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(CONF.network.tenant_network_cidr)
        mask_bits = CONF.network.tenant_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=4)
                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=4)
                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):
        external_net_id = dict(
            network_id=CONF.network.public_network_id)
        resp_body = self.network_admin_client.create_router(
            router_name,
            external_gateway_info=external_net_id,
            tenant_id=tenant_id)
        return resp_body['router']

    def _add_router_interface(self, router_id, subnet_id):
        self.network_admin_client.add_router_interface(router_id,
                                                       subnet_id=subnet_id)

    def get_credentials(self, credential_type):
        if self._creds.get(str(credential_type)):
            credentials = self._creds[str(credential_type)]
        else:
            if 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)
            self._creds[str(credential_type)] = credentials
            # Maintained until tests are ported
            LOG.info("Acquired dynamic creds:\n credentials: %s"
                     % credentials)
            if (CONF.service_available.neutron and
                not CONF.baremetal.driver_enabled and
                CONF.auth.create_isolated_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)
        return credentials

    def get_primary_creds(self):
        return self.get_credentials('primary')

    def get_admin_creds(self):
        return self.get_credentials('admin')

    def get_alt_creds(self):
        return self.get_credentials('alt')

    def get_creds_by_roles(self, roles, force_new=False):
        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.
        exist_creds = self._creds.get(str(roles))
        # 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 = str(roles) + '-' + str(len(self._creds))
            self._creds[new_index] = exist_creds
            del self._creds[str(roles)]
        return self.get_credentials(roles)

    def _clear_isolated_router(self, router_id, router_name):
        net_client = self.network_admin_client
        try:
            net_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 _cleanup_default_secgroup(self, tenant):
        nsg_client = self.security_groups_admin_client
        resp_body = nsg_client.list_security_groups(tenant_id=tenant,
                                                    name="default")
        secgroups_to_delete = resp_body['security_groups']
        for secgroup in secgroups_to_delete:
            try:
                nsg_client.delete_security_group(secgroup['id'])
            except lib_exc.NotFound:
                LOG.warning('Security group %s, id %s not found for clean-up' %
                            (secgroup['name'], secgroup['id']))

    def _clear_isolated_net_resources(self):
        net_client = self.network_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:
                    net_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()
        for creds in six.itervalues(self._creds):
            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)
            try:
                if CONF.service_available.neutron:
                    self._cleanup_default_secgroup(creds.tenant_id)
                self.creds_client.delete_project(creds.tenant_id)
            except lib_exc.NotFound:
                LOG.warning("tenant with name: %s not found for delete" %
                            creds.tenant_name)
        self._creds = {}

    def is_multi_user(self):
        return True

    def is_multi_tenant(self):
        return True

    def is_role_available(self, role):
        return True