text: # Copyright 2013 Red Hat, Inc.
  # All Rights Reserved.
  #
  #    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 oslo_log import log
  import testtools
  
  from tempest import clients
  from tempest.common.utils import data_utils
  from tempest import config
  from tempest.scenario import manager
  from tempest import test
  
  CONF = config.CONF
  LOG = log.getLogger(__name__)
  
  
  class TestSecurityGroupsBasicOps(manager.NetworkScenarioTest):
  
      """The test suite for security groups
  
      This test suite assumes that Nova has been configured to
      boot VM's with Neutron-managed networking, and attempts to
      verify cross tenant connectivity as follows
  
      ssh:
          in order to overcome "ip namespace", each tenant has an "access point"
          VM with floating-ip open to incoming ssh connection allowing network
          commands (ping/ssh) to be executed from within the
          tenant-network-namespace
          Tempest host performs key-based authentication to the ssh server via
          floating IP address
  
      connectivity test is done by pinging destination server via source server
      ssh connection.
      success - ping returns
      failure - ping_timeout reached
  
      multi-node:
          Multi-Node mode is enabled when CONF.compute.min_compute_nodes > 1.
          Tests connectivity between servers on different compute nodes.
          When enabled, test will boot each new server to different
          compute nodes.
  
      setup:
          for primary tenant:
              1. create a network&subnet
              2. create a router (if public router isn't configured)
              3. connect tenant network to public network via router
              4. create an access point:
                  a. a security group open to incoming ssh connection
                  b. a VM with a floating ip
              5. create a general empty security group (same as "default", but
              without rules allowing in-tenant traffic)
  
      tests:
          1. _verify_network_details
          2. _verify_mac_addr: for each access point verify that
          (subnet, fix_ip, mac address) are as defined in the port list
          3. _test_in_tenant_block: test that in-tenant traffic is disabled
          without rules allowing it
          4. _test_in_tenant_allow: test that in-tenant traffic is enabled
          once an appropriate rule has been created
          5. _test_cross_tenant_block: test that cross-tenant traffic is disabled
          without a rule allowing it on destination tenant
          6. _test_cross_tenant_allow:
              * test that cross-tenant traffic is enabled once an appropriate
              rule has been created on destination tenant.
              * test that reverse traffic is still blocked
              * test than reverse traffic is enabled once an appropriate rule has
              been created on source tenant
          7._test_port_update_new_security_group:
             * test that traffic is blocked with default security group
             * test that traffic is enabled after updating port with new security
             group having appropriate rule
          8. _test_multiple_security_groups: test multiple security groups can be
             associated with the vm
  
      assumptions:
          1. alt_tenant/user existed and is different from primary_tenant/user
          2. Public network is defined and reachable from the Tempest host
          3. Public router can either be:
              * defined, in which case all tenants networks can connect directly
              to it, and cross tenant check will be done on the private IP of the
              destination tenant
              or
              * not defined (empty string), in which case each tenant will have
              its own router connected to the public network
      """
  
      credentials = ['primary', 'alt', 'admin']
  
      class TenantProperties(object):
          """helper class to save tenant details
  
              id
              credentials
              network
              subnet
              security groups
              servers
              access point
          """
  
          def __init__(self, credentials):
              self.manager = clients.Manager(credentials)
              # Credentials from manager are filled with both names and IDs
              self.creds = self.manager.credentials
              self.network = None
              self.subnet = None
              self.router = None
              self.security_groups = {}
              self.servers = list()
  
          def set_network(self, network, subnet, router):
              self.network = network
              self.subnet = subnet
              self.router = router
  
      @classmethod
      def skip_checks(cls):
          super(TestSecurityGroupsBasicOps, cls).skip_checks()
          if CONF.baremetal.driver_enabled:
              msg = ('Not currently supported by baremetal.')
              raise cls.skipException(msg)
          if CONF.network.port_vnic_type in ['direct', 'macvtap']:
              msg = ('Not currently supported when using vnic_type'
                     ' direct or macvtap')
              raise cls.skipException(msg)
          if not (CONF.network.project_networks_reachable or
                  CONF.network.public_network_id):
              msg = ('Either project_networks_reachable must be "true", or '
                     'public_network_id must be defined.')
              raise cls.skipException(msg)
          if not test.is_extension_enabled('security-group', 'network'):
              msg = "security-group extension not enabled."
              raise cls.skipException(msg)
  
      @classmethod
      def setup_credentials(cls):
          # Create no network resources for these tests.
          cls.set_network_resources()
          super(TestSecurityGroupsBasicOps, cls).setup_credentials()
          # TODO(mnewby) Consider looking up entities as needed instead
          # of storing them as collections on the class.
  
          # Credentials from the manager are filled with both IDs and Names
          cls.alt_creds = cls.alt_manager.credentials
  
      @classmethod
      def resource_setup(cls):
          super(TestSecurityGroupsBasicOps, cls).resource_setup()
  
          cls.multi_node = CONF.compute.min_compute_nodes > 1 and \
              test.is_scheduler_filter_enabled("DifferentHostFilter")
          if cls.multi_node:
              LOG.info("Working in Multi Node mode")
          else:
              LOG.info("Working in Single Node mode")
  
          cls.floating_ips = {}
          cls.tenants = {}
          creds = cls.manager.credentials
          cls.primary_tenant = cls.TenantProperties(creds)
          cls.alt_tenant = cls.TenantProperties(cls.alt_creds)
          for tenant in [cls.primary_tenant, cls.alt_tenant]:
              cls.tenants[tenant.creds.tenant_id] = tenant
  
          cls.floating_ip_access = not CONF.network.public_router_id
  
      def setUp(self):
          """Set up a single tenant with an accessible server.
  
          If multi-host is enabled, save created server uuids.
          """
          self.servers = []
  
          super(TestSecurityGroupsBasicOps, self).setUp()
          self._deploy_tenant(self.primary_tenant)
          self._verify_network_details(self.primary_tenant)
          self._verify_mac_addr(self.primary_tenant)
  
      def _create_tenant_keypairs(self, tenant):
          keypair = self.create_keypair(tenant.manager.keypairs_client)
          tenant.keypair = keypair
  
      def _create_tenant_security_groups(self, tenant):
          access_sg = self._create_empty_security_group(
              namestart='secgroup_access-',
              tenant_id=tenant.creds.tenant_id,
              client=tenant.manager.security_groups_client
          )
  
          # don't use default secgroup since it allows in-project traffic
          def_sg = self._create_empty_security_group(
              namestart='secgroup_general-',
              tenant_id=tenant.creds.tenant_id,
              client=tenant.manager.security_groups_client
          )
          tenant.security_groups.update(access=access_sg, default=def_sg)
          ssh_rule = dict(
              protocol='tcp',
              port_range_min=22,
              port_range_max=22,
              direction='ingress',
          )
          sec_group_rules_client = tenant.manager.security_group_rules_client
          self._create_security_group_rule(
              secgroup=access_sg,
              sec_group_rules_client=sec_group_rules_client,
              **ssh_rule)
  
      def _verify_network_details(self, tenant):
          # Checks that we see the newly created network/subnet/router via
          # checking the result of list_[networks,routers,subnets]
          # Check that (router, subnet) couple exist in port_list
          seen_nets = self._list_networks()
          seen_names = [n['name'] for n in seen_nets]
          seen_ids = [n['id'] for n in seen_nets]
  
          self.assertIn(tenant.network['name'], seen_names)
          self.assertIn(tenant.network['id'], seen_ids)
  
          seen_subnets = [(n['id'], n['cidr'], n['network_id'])
                          for n in self._list_subnets()]
          mysubnet = (tenant.subnet['id'], tenant.subnet['cidr'],
                      tenant.network['id'])
          self.assertIn(mysubnet, seen_subnets)
  
          seen_routers = self._list_routers()
          seen_router_ids = [n['id'] for n in seen_routers]
          seen_router_names = [n['name'] for n in seen_routers]
  
          self.assertIn(tenant.router['name'], seen_router_names)
          self.assertIn(tenant.router['id'], seen_router_ids)
  
          myport = (tenant.router['id'], tenant.subnet['id'])
          router_ports = [(i['device_id'], i['fixed_ips'][0]['subnet_id']) for i
                          in self._list_ports()
                          if self._is_router_port(i)]
  
          self.assertIn(myport, router_ports)
  
      def _is_router_port(self, port):
          """Return True if port is a router interface."""
          # NOTE(armando-migliaccio): match device owner for both centralized
          # and distributed routers; 'device_owner' is "" by default.
          return port['device_owner'].startswith('network:router_interface')
  
      def _create_server(self, name, tenant, security_groups, **kwargs):
          """Creates a server and assigns it to security group.
  
          If multi-host is enabled, Ensures servers are created on different
          compute nodes, by storing created servers' ids and uses different_host
          as scheduler_hints on creation.
          Validates servers are created as requested, using admin client.
          """
          security_groups_names = [{'name': s['name']} for s in security_groups]
          if self.multi_node:
              kwargs["scheduler_hints"] = {'different_host': self.servers}
          server = self.create_server(
              name=name,
              networks=[{'uuid': tenant.network["id"]}],
              key_name=tenant.keypair['name'],
              security_groups=security_groups_names,
              wait_until='ACTIVE',
              clients=tenant.manager,
              **kwargs)
          if 'security_groups' in server:
              self.assertEqual(
                  sorted([s['name'] for s in security_groups]),
                  sorted([s['name'] for s in server['security_groups']]))
  
          # Verify servers are on different compute nodes
          if self.multi_node:
              adm_get_server = self.admin_manager.servers_client.show_server
              new_host = adm_get_server(server["id"])["server"][
                  "OS-EXT-SRV-ATTR:host"]
              host_list = [adm_get_server(s)["server"]["OS-EXT-SRV-ATTR:host"]
                           for s in self.servers]
              self.assertNotIn(new_host, host_list,
                               message="Failed to boot servers on different "
                                       "Compute nodes.")
  
              self.servers.append(server["id"])
  
          return server
  
      def _create_tenant_servers(self, tenant, num=1):
          for i in range(num):
              name = 'server-{tenant}-gen-{num}'.format(
                     tenant=tenant.creds.tenant_name,
                     num=i
              )
              name = data_utils.rand_name(name)
              server = self._create_server(name, tenant,
                                           [tenant.security_groups['default']])
              tenant.servers.append(server)
  
      def _set_access_point(self, tenant):
          # creates a server in a secgroup with rule allowing external ssh
          # in order to access project internal network
          # workaround ip namespace
          secgroups = tenant.security_groups.values()
          name = 'server-{tenant}-access_point'.format(
              tenant=tenant.creds.tenant_name)
          name = data_utils.rand_name(name)
          server = self._create_server(name, tenant,
                                       security_groups=secgroups)
          tenant.access_point = server
          self._assign_floating_ips(tenant, server)
  
      def _assign_floating_ips(self, tenant, server):
          public_network_id = CONF.network.public_network_id
          floating_ip = self.create_floating_ip(
              server, public_network_id,
              client=tenant.manager.floating_ips_client)
          self.floating_ips.setdefault(server['id'], floating_ip)
  
      def _create_tenant_network(self, tenant, port_security_enabled=True):
          network, subnet, router = self.create_networks(
              networks_client=tenant.manager.networks_client,
              routers_client=tenant.manager.routers_client,
              subnets_client=tenant.manager.subnets_client,
              port_security_enabled=port_security_enabled)
          tenant.set_network(network, subnet, router)
  
      def _deploy_tenant(self, tenant_or_id):
          """creates:
  
              network
              subnet
              router (if public not defined)
              access security group
              access-point server
          """
          if not isinstance(tenant_or_id, self.TenantProperties):
              tenant = self.tenants[tenant_or_id]
          else:
              tenant = tenant_or_id
          self._create_tenant_keypairs(tenant)
          self._create_tenant_network(tenant)
          self._create_tenant_security_groups(tenant)
          self._set_access_point(tenant)
  
      def _get_server_ip(self, server, floating=False):
          """returns the ip (floating/internal) of a server"""
          if floating:
              server_ip = self.floating_ips[server['id']]['floating_ip_address']
          else:
              server_ip = None
              network_name = self.tenants[server['tenant_id']].network['name']
              if network_name in server['addresses']:
                  server_ip = server['addresses'][network_name][0]['addr']
          return server_ip
  
      def _connect_to_access_point(self, tenant):
          """create ssh connection to tenant access point"""
          access_point_ssh = \
              self.floating_ips[tenant.access_point['id']]['floating_ip_address']
          private_key = tenant.keypair['private_key']
          access_point_ssh = self.get_remote_client(
              access_point_ssh, private_key=private_key)
          return access_point_ssh
  
      def _check_connectivity(self, access_point, ip, should_succeed=True):
          if should_succeed:
              msg = "Timed out waiting for %s to become reachable" % ip
          else:
              msg = "%s is reachable" % ip
          self.assertTrue(self._check_remote_connectivity(access_point, ip,
                                                          should_succeed), msg)
  
      def _test_in_tenant_block(self, tenant):
          access_point_ssh = self._connect_to_access_point(tenant)
          for server in tenant.servers:
              self._check_connectivity(access_point=access_point_ssh,
                                       ip=self._get_server_ip(server),
                                       should_succeed=False)
  
      def _test_in_tenant_allow(self, tenant):
          ruleset = dict(
              protocol='icmp',
              remote_group_id=tenant.security_groups['default']['id'],
              direction='ingress'
          )
          self._create_security_group_rule(
              secgroup=tenant.security_groups['default'],
              security_groups_client=tenant.manager.security_groups_client,
              **ruleset
          )
          access_point_ssh = self._connect_to_access_point(tenant)
          for server in tenant.servers:
              self._check_connectivity(access_point=access_point_ssh,
                                       ip=self._get_server_ip(server))
  
      def _test_cross_tenant_block(self, source_tenant, dest_tenant):
          # if public router isn't defined, then dest_tenant access is via
          # floating-ip
          access_point_ssh = self._connect_to_access_point(source_tenant)
          ip = self._get_server_ip(dest_tenant.access_point,
                                   floating=self.floating_ip_access)
          self._check_connectivity(access_point=access_point_ssh, ip=ip,
                                   should_succeed=False)
  
      def _test_cross_tenant_allow(self, source_tenant, dest_tenant):
          """check for each direction:
  
          creating rule for tenant incoming traffic enables only 1way traffic
          """
          ruleset = dict(
              protocol='icmp',
              direction='ingress'
          )
          sec_group_rules_client = (
              dest_tenant.manager.security_group_rules_client)
          self._create_security_group_rule(
              secgroup=dest_tenant.security_groups['default'],
              sec_group_rules_client=sec_group_rules_client,
              **ruleset
          )
          access_point_ssh = self._connect_to_access_point(source_tenant)
          ip = self._get_server_ip(dest_tenant.access_point,
                                   floating=self.floating_ip_access)
          self._check_connectivity(access_point_ssh, ip)
  
          # test that reverse traffic is still blocked
          self._test_cross_tenant_block(dest_tenant, source_tenant)
  
          # allow reverse traffic and check
          sec_group_rules_client = (
              source_tenant.manager.security_group_rules_client)
          self._create_security_group_rule(
              secgroup=source_tenant.security_groups['default'],
              sec_group_rules_client=sec_group_rules_client,
              **ruleset
          )
  
          access_point_ssh_2 = self._connect_to_access_point(dest_tenant)
          ip = self._get_server_ip(source_tenant.access_point,
                                   floating=self.floating_ip_access)
          self._check_connectivity(access_point_ssh_2, ip)
  
      def _verify_mac_addr(self, tenant):
          """Verify that VM has the same ip, mac as listed in port"""
  
          access_point_ssh = self._connect_to_access_point(tenant)
          mac_addr = access_point_ssh.get_mac_address()
          mac_addr = mac_addr.strip().lower()
          # Get the fixed_ips and mac_address fields of all ports. Select
          # only those two columns to reduce the size of the response.
          port_list = self._list_ports(fields=['fixed_ips', 'mac_address'])
          port_detail_list = [
              (port['fixed_ips'][0]['subnet_id'],
               port['fixed_ips'][0]['ip_address'],
               port['mac_address'].lower())
              for port in port_list if port['fixed_ips']
          ]
          server_ip = self._get_server_ip(tenant.access_point)
          subnet_id = tenant.subnet['id']
          self.assertIn((subnet_id, server_ip, mac_addr), port_detail_list)
  
      @test.idempotent_id('e79f879e-debb-440c-a7e4-efeda05b6848')
      @test.services('compute', 'network')
      def test_cross_tenant_traffic(self):
          if not self.credentials_provider.is_multi_tenant():
              raise self.skipException("No secondary tenant defined")
          try:
              # deploy new project
              self._deploy_tenant(self.alt_tenant)
              self._verify_network_details(self.alt_tenant)
              self._verify_mac_addr(self.alt_tenant)
  
              # cross tenant check
              source_tenant = self.primary_tenant
              dest_tenant = self.alt_tenant
              self._test_cross_tenant_block(source_tenant, dest_tenant)
              self._test_cross_tenant_allow(source_tenant, dest_tenant)
          except Exception:
              for tenant in self.tenants.values():
                  self._log_console_output(servers=tenant.servers)
              raise
  
      @test.idempotent_id('63163892-bbf6-4249-aa12-d5ea1f8f421b')
      @test.services('compute', 'network')
      def test_in_tenant_traffic(self):
          try:
              self._create_tenant_servers(self.primary_tenant, num=1)
  
              # in-tenant check
              self._test_in_tenant_block(self.primary_tenant)
              self._test_in_tenant_allow(self.primary_tenant)
          except Exception:
              for tenant in self.tenants.values():
                  self._log_console_output(servers=tenant.servers)
              raise
  
      @test.idempotent_id('f4d556d7-1526-42ad-bafb-6bebf48568f6')
      @test.services('compute', 'network')
      def test_port_update_new_security_group(self):
          """Verifies the traffic after updating the vm port
  
          With new security group having appropriate rule.
          """
          new_tenant = self.primary_tenant
  
          # Create empty security group and add icmp rule in it
          new_sg = self._create_empty_security_group(
              namestart='secgroup_new-',
              tenant_id=new_tenant.creds.tenant_id,
              client=new_tenant.manager.security_groups_client)
          icmp_rule = dict(
              protocol='icmp',
              direction='ingress',
          )
          sec_group_rules_client = new_tenant.manager.security_group_rules_client
          self._create_security_group_rule(
              secgroup=new_sg,
              sec_group_rules_client=sec_group_rules_client,
              **icmp_rule)
          new_tenant.security_groups.update(new_sg=new_sg)
  
          # Create server with default security group
          name = 'server-{tenant}-gen-1'.format(
                 tenant=new_tenant.creds.tenant_name
          )
          name = data_utils.rand_name(name)
          server = self._create_server(name, new_tenant,
                                       [new_tenant.security_groups['default']])
  
          # Check connectivity failure with default security group
          try:
              access_point_ssh = self._connect_to_access_point(new_tenant)
              self._check_connectivity(access_point=access_point_ssh,
                                       ip=self._get_server_ip(server),
                                       should_succeed=False)
              server_id = server['id']
              port_id = self._list_ports(device_id=server_id)[0]['id']
  
              # update port with new security group and check connectivity
              self.ports_client.update_port(port_id, security_groups=[
                  new_tenant.security_groups['new_sg']['id']])
              self._check_connectivity(
                  access_point=access_point_ssh,
                  ip=self._get_server_ip(server))
          except Exception:
              for tenant in self.tenants.values():
                  self._log_console_output(servers=tenant.servers)
              raise
  
      @test.idempotent_id('d2f77418-fcc4-439d-b935-72eca704e293')
      @test.services('compute', 'network')
      def test_multiple_security_groups(self):
          """Verify multiple security groups and checks that rules
  
          provided in the both the groups is applied onto VM
          """
          tenant = self.primary_tenant
          ip = self._get_server_ip(tenant.access_point,
                                   floating=self.floating_ip_access)
          ssh_login = CONF.validation.image_ssh_user
          private_key = tenant.keypair['private_key']
          self.check_vm_connectivity(ip,
                                     should_connect=False)
          ruleset = dict(
              protocol='icmp',
              direction='ingress'
          )
          self._create_security_group_rule(
              secgroup=tenant.security_groups['default'],
              **ruleset
          )
          # NOTE: Vm now has 2 security groups one with ssh rule(
          # already added in setUp() method),and other with icmp rule
          # (added in the above step).The check_vm_connectivity tests
          # -that vm ping test is successful
          # -ssh to vm is successful
          self.check_vm_connectivity(ip,
                                     username=ssh_login,
                                     private_key=private_key,
                                     should_connect=True)
  
      @test.requires_ext(service='network', extension='port-security')
      @test.idempotent_id('7c811dcc-263b-49a3-92d2-1b4d8405f50c')
      @test.services('compute', 'network')
      def test_port_security_disable_security_group(self):
          """Verify the default security group rules is disabled."""
          new_tenant = self.primary_tenant
  
          # Create server
          name = 'server-{tenant}-gen-1'.format(
                 tenant=new_tenant.creds.tenant_name
          )
          name = data_utils.rand_name(name)
          server = self._create_server(name, new_tenant,
                                       [new_tenant.security_groups['default']])
  
          access_point_ssh = self._connect_to_access_point(new_tenant)
          server_id = server['id']
          port_id = self._list_ports(device_id=server_id)[0]['id']
  
          # Flip the port's port security and check connectivity
          try:
              self.ports_client.update_port(port_id,
                                            port_security_enabled=True,
                                            security_groups=[])
              self._check_connectivity(access_point=access_point_ssh,
                                       ip=self._get_server_ip(server),
                                       should_succeed=False)
  
              self.ports_client.update_port(port_id,
                                            port_security_enabled=False,
                                            security_groups=[])
              self._check_connectivity(
                  access_point=access_point_ssh,
                  ip=self._get_server_ip(server))
          except Exception:
              for tenant in self.tenants.values():
                  self._log_console_output(servers=tenant.servers)
              raise
  
      @test.requires_ext(service='network', extension='port-security')
      @test.idempotent_id('13ccf253-e5ad-424b-9c4a-97b88a026699')
      @testtools.skipUnless(
          CONF.compute_feature_enabled.allow_port_security_disabled,
          'Port security must be enabled.')
      # TODO(mriedem): We shouldn't actually need to check this since neutron
      # disables the port_security extension by default, but the problem is nova
      # assumes port_security_enabled=True if it's not set on the network
      # resource, which will mean nova may attempt to apply a security group on
      # a port on that network which would fail. This is really a bug in nova.
      @testtools.skipUnless(
          CONF.network_feature_enabled.port_security,
          'Port security must be enabled.')
      @test.services('compute', 'network')
      def test_boot_into_disabled_port_security_network_without_secgroup(self):
          tenant = self.primary_tenant
          self._create_tenant_network(tenant, port_security_enabled=False)
          self.assertFalse(tenant.network['port_security_enabled'])
          name = data_utils.rand_name('server-smoke')
          sec_groups = []
          server = self._create_server(name, tenant, sec_groups)
          server_id = server['id']
          ports = self._list_ports(device_id=server_id)
          self.assertEqual(1, len(ports))
          for port in ports:
              self.assertEmpty(port['security_groups'],
                               "Neutron shouldn't even use it's default sec "
                               "group.")
  