text: # -*- encoding: utf-8 -*-
  #
  #    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.
  """
  Tests for ACL. Checks whether certain kinds of requests
  are blocked or allowed to be processed.
  """
  
  import abc
  from unittest import mock
  
  import ddt
  from keystonemiddleware import auth_token
  from oslo_config import cfg
  
  from ironic.api.controllers.v1 import versions as api_versions
  from ironic.common import exception
  from ironic.conductor import rpcapi
  from ironic.db import api as db_api
  from ironic.tests.unit.api import base
  from ironic.tests.unit.db import utils as db_utils
  
  
  class TestACLBase(base.BaseApiTest):
  
      def setUp(self):
          super(TestACLBase, self).setUp()
  
          self.environ = {}
          self.format_data = {}
          self._create_test_data()
          self.fake_token = None
          mock_auth = mock.patch.object(
              auth_token.AuthProtocol, 'process_request',
              autospec=True)
          self.mock_auth = mock_auth.start()
          self.addCleanup(mock_auth.stop)
  
          topic = mock.patch.object(
              rpcapi.ConductorAPI, 'get_topic_for', autospec=True)
          self.mock_topic = topic.start()
          self.mock_topic.side_effect = exception.TemporaryFailure
          self.addCleanup(topic.stop)
          rtopic = mock.patch.object(rpcapi.ConductorAPI, 'get_random_topic',
                                     autospec=True)
          self.mock_random_topic = rtopic.start()
          self.mock_random_topic.side_effect = exception.TemporaryFailure
          self.addCleanup(rtopic.stop)
          self._set_test_config()
  
      def _make_app(self):
          cfg.CONF.set_override('auth_strategy', 'keystone')
          return super(TestACLBase, self)._make_app()
  
      @abc.abstractmethod
      def _create_test_data(self):
          pass
  
      @abc.abstractmethod
      def _set_test_config(self):
          pass
  
      def _check_skip(self, **kwargs):
          if kwargs.get('skip_reason'):
              self.skipTest(kwargs.get('skip_reason'))
  
      def _fake_process_request(self, request, auth_token_request):
          pass
  
      def _test_request(self, path, params=None, headers=None, method='get',
                        body=None, assert_status=None,
                        assert_dict_contains=None,
                        assert_list_length=None,
                        deprecated=None):
          path = path.format(**self.format_data)
          self.mock_auth.side_effect = self._fake_process_request
  
          # always request the latest api version
          version = api_versions.max_version_string()
          rheaders = {
              'X-OpenStack-Ironic-API-Version': version
          }
          # NOTE(TheJulia): Logging the test request to aid
          # in troubleshooting ACL testing. This is a pattern
          # followed in API unit testing in ironic, and
          # really does help.
          print('API ACL Testing Path %s %s' % (method, path))
          if headers:
              for k, v in headers.items():
                  rheaders[k] = v.format(**self.format_data)
          if method == 'get':
              response = self.get_json(
                  path,
                  headers=rheaders,
                  expect_errors=True,
                  extra_environ=self.environ,
                  path_prefix=''
              )
          elif method == 'put':
              response = self.put_json(
                  path,
                  headers=rheaders,
                  expect_errors=True,
                  extra_environ=self.environ,
                  path_prefix='',
                  params=body
              )
          elif method == 'post':
              response = self.post_json(
                  path,
                  headers=rheaders,
                  expect_errors=True,
                  extra_environ=self.environ,
                  path_prefix='',
                  params=body
              )
          elif method == 'patch':
              response = self.patch_json(
                  path,
                  params=body,
                  headers=rheaders,
                  expect_errors=True,
                  extra_environ=self.environ,
                  path_prefix=''
              )
          elif method == 'delete':
              response = self.delete(
                  path,
                  headers=rheaders,
                  expect_errors=True,
                  extra_environ=self.environ,
                  path_prefix=''
              )
          else:
              assert False, 'Unimplemented test method: %s' % method
          # Once miggrated:
          # Items will return:
          # 403 - Trying to access something that is generally denied.
          #       Example: PATCH /v1/nodes/<uuid> as a reader.
          # 404 - Trying to access something where we don't have permissions
          #       in a project scope. This is particularly true where implied
          #       permissions or assocation exists. Ports are attempted to be
          #       accessed when the underlying node is inaccessible as owner
          #       nor node matches.
          #       Example: GET /v1/portgroups or /v1/nodes/<uuid>/ports
          # 500 - Attempting to access something such an system scoped endpoint
          #       with a project scoped request. Example: /v1/conductors.
          if not (bool(deprecated)
                  and ('404' in response.status
                       or '500' in response.status
                       or '403' in response.status)
                  and cfg.CONF.oslo_policy.enforce_scope
                  and cfg.CONF.oslo_policy.enforce_new_defaults):
              self.assertEqual(assert_status, response.status_int)
          else:
              self.assertTrue(
                  ('404' in response.status
                   or '500' in response.status
                   or '403' in response.status))
              # We can't check the contents of the response if there is no
              # response.
              return
          if not bool(deprecated):
              self.assertIsNotNone(assert_status,
                                   'Tests must include an assert_status')
  
          if assert_dict_contains:
              for k, v in assert_dict_contains.items():
                  self.assertIn(k, response)
                  print(k)
                  print(v)
                  if str(v) == "None":
                      # Compare since the variable loaded from the
                      # json ends up being null in json or None.
                      self.assertIsNone(response.json[k])
                  elif str(v) == "{}":
                      # Special match for signifying a dictonary.
                      self.assertEqual({}, response.json[k])
                  elif isinstance(v, dict):
                      # The value from the YAML can be a dictionary,
                      # which cannot be formatted, so we're likely doing
                      # direct matching.
                      self.assertEqual(str(v), str(response.json[k]))
                  else:
                      self.assertEqual(v.format(**self.format_data),
                                       response.json[k])
  
          if assert_list_length:
              for root, length in assert_list_length.items():
                  # root - object to look inside
                  # length - number of expected elements which will be
                  #          important for owner/lessee testing.
                  items = response.json[root]
                  self.assertIsInstance(items, list)
                  if not (bool(deprecated)
                          and cfg.CONF.oslo_policy.enforce_scope):
                      self.assertEqual(length, len(items))
                  else:
                      # If we have scope enforcement, we likely have different
                      # views, such as "other" admins being subjected to
                      # a filtered view in these cases.
                      self.assertEqual(0, len(items))
  
          # NOTE(TheJulia): API tests in Ironic tend to have a pattern
          # to print request and response data to aid in development
          # and troubleshooting. As such the prints should remain,
          # at least until we are through primary development of the
          # this test suite.
          print('ACL Test GOT %s' % response)
  
  
  @ddt.ddt
  class TestRBACBasic(TestACLBase):
  
      def _create_test_data(self):
          fake_db_node = db_utils.create_test_node(chassis_id=None)
          self.format_data['node_uuid'] = fake_db_node['uuid']
  
      @ddt.file_data('test_acl_basic.yaml')
      @ddt.unpack
      def test_basic(self, **kwargs):
          self._check_skip(**kwargs)
          self._test_request(**kwargs)
  
  
  @ddt.ddt
  class TestRBACModelBeforeScopesBase(TestACLBase):
  
      def _create_test_data(self):
          allocated_node_id = 31
          fake_db_allocation = db_utils.create_test_allocation(
              resource_class="CUSTOM_TEST")
          fake_db_node = db_utils.create_test_node(
              chassis_id=None,
              driver='fake-driverz',
              owner='z')
          fake_db_node_alloced = db_utils.create_test_node(
              id=allocated_node_id,
              chassis_id=None,
              uuid='22e26c0b-03f2-4d2e-ae87-c02d7f33c000',
              driver='fake-driverz',
              owner='z')
          dbapi = db_api.get_instance()
          dbapi.update_allocation(fake_db_allocation['id'],
                                  dict(node_id=allocated_node_id))
          fake_vif_port_id = "ee21d58f-5de2-4956-85ff-33935ea1ca00"
          fake_db_port = db_utils.create_test_port(
              node_id=fake_db_node['id'],
              internal_info={'tenant_vif_port_id': fake_vif_port_id})
          fake_db_portgroup = db_utils.create_test_portgroup(
              uuid="6eb02b44-18a3-4659-8c0b-8d2802581ae4",
              node_id=fake_db_node['id'])
          fake_db_chassis = db_utils.create_test_chassis(
              drivers=['fake-hardware', 'fake-driverz', 'fake-driver'])
          fake_db_deploy_template = db_utils.create_test_deploy_template()
          fake_db_conductor = db_utils.create_test_conductor()
          fake_db_volume_target = db_utils.create_test_volume_target(
              node_id=fake_db_node['id'])
          fake_db_volume_connector = db_utils.create_test_volume_connector(
              node_id=fake_db_node['id'])
          # Trait name aligns with create_test_node_trait.
          fake_trait = 'trait'
          fake_setting = 'FAKE_SETTING'
          db_utils.create_test_bios_setting(
              node_id=fake_db_node['id'],
              name=fake_setting,
              value=fake_setting)
          db_utils.create_test_node_trait(
              node_id=fake_db_node['id'])
          fake_history = db_utils.create_test_history(node_id=fake_db_node.id)
          # dedicated node for portgroup addition test to avoid
          # false positives with test runners.
          db_utils.create_test_node(
              uuid='18a552fb-dcd2-43bf-9302-e4c93287be11')
          self.format_data.update({
              'node_ident': fake_db_node['uuid'],
              'allocated_node_ident': fake_db_node_alloced['uuid'],
              'port_ident': fake_db_port['uuid'],
              'portgroup_ident': fake_db_portgroup['uuid'],
              'chassis_ident': fake_db_chassis['uuid'],
              'deploy_template_ident': fake_db_deploy_template['uuid'],
              'allocation_ident': fake_db_allocation['uuid'],
              'conductor_ident': fake_db_conductor['hostname'],
              'vif_ident': fake_vif_port_id,
              # Can't use the same fake-driver as other tests can
              # pollute a global method cache in the API that is in the
              # test runner, resulting in false positives.
              'driver_name': 'fake-driverz',
              'bios_setting': fake_setting,
              'trait': fake_trait,
              'volume_target_ident': fake_db_volume_target['uuid'],
              'volume_connector_ident': fake_db_volume_connector['uuid'],
              'history_ident': fake_history['uuid'],
          })
  
  
  @ddt.ddt
  class TestRBACModelBeforeScopes(TestRBACModelBeforeScopesBase):
  
      def _set_test_config(self):
          # NOTE(TheJulia): Sets default test conditions, in the event
          # oslo_policy defaults change.
          cfg.CONF.set_override('enforce_scope', False, group='oslo_policy')
          cfg.CONF.set_override('enforce_new_defaults', False,
                                group='oslo_policy')
  
      @ddt.file_data('test_rbac_legacy.yaml')
      @ddt.unpack
      def test_rbac_legacy(self, **kwargs):
          self._check_skip(**kwargs)
          self._test_request(**kwargs)
  
  
  @ddt.ddt
  class TestRBACScoped(TestRBACModelBeforeScopes):
      """Test Scoped RBAC access using our existing access policy."""
  
      def _set_test_config(self):
          # NOTE(TheJulia): This test class is as like a canary.
          # The operational intent is for it to kind of provide
          # a safety net as we're changing policy rules so we can
          # incremently disable the ones we *know* will no longer work
          # while we also enable the new ones in another test class with
          # the appropriate scope friendly chagnges. In other words, two
          # test changes will be needed for each which should also reduce
          # risk of accidential policy changes. It may just be Julia being
          # super risk-adverse, just let her roll with it and we will delete
          # this class later.
          # NOTE(TheJulia): This test class runs with test_rbac_legacy.yaml!
          cfg.CONF.set_override('enforce_scope', True, group='oslo_policy')
          cfg.CONF.set_override('enforce_new_defaults', True,
                                group='oslo_policy')
  
      @ddt.file_data('test_rbac_legacy.yaml')
      def test_scoped_canary(self, **kwargs):
          self._check_skip(**kwargs)
          self._test_request(**kwargs)
  
  
  @ddt.ddt
  class TestRBACScopedRequests(TestRBACModelBeforeScopesBase):
  
      @ddt.file_data('test_rbac_system_scoped.yaml')
      @ddt.unpack
      def test_system_scoped(self, **kwargs):
          self._check_skip(**kwargs)
          self._test_request(**kwargs)
  
  
  @ddt.ddt
  class TestRBACProjectScoped(TestACLBase):
  
      def setUp(self):
          super(TestRBACProjectScoped, self).setUp()
  
          cfg.CONF.set_override('enforce_scope', True, group='oslo_policy')
          cfg.CONF.set_override('enforce_new_defaults', True,
                                group='oslo_policy')
  
      def _create_test_data(self):
          owner_node_ident = '1ab63b9e-66d7-4cd7-8618-dddd0f9f7881'
          lessee_node_ident = '38d5abed-c585-4fce-a57e-a2ffc2a2ec6f'
          owner_project_id = '70e5e25a-2ca2-4cb1-8ae8-7d8739cee205'
          lessee_project_id = 'f11853c7-fa9c-4db3-a477-c9d8e0dbbf13'
          unowned_node = db_utils.create_test_node(chassis_id=None)
  
          # owned node - since the tests use the same node for
          # owner/lesse checks
          owned_node = db_utils.create_test_node(
              uuid=owner_node_ident,
              owner=owner_project_id,
              last_error='meow',
              reservation='lolcats')
          owned_node_port = db_utils.create_test_port(
              uuid='ebe30f19-358d-41e1-8d28-fd7357a0164c',
              node_id=owned_node['id'],
              address='00:00:00:00:00:01')
          db_utils.create_test_port(
              uuid='21a3c5a7-1e14-44dc-a9dd-0c84d5477a57',
              node_id=owned_node['id'],
              address='00:00:00:00:00:02')
          owner_pgroup = db_utils.create_test_portgroup(
              uuid='b16efcf3-2990-41a1-bc1d-5e2c16f3d5fc',
              node_id=owned_node['id'],
              name='magicfoo',
              address='01:03:09:ff:01:01')
          db_utils.create_test_volume_target(
              uuid='a265e2f0-e97f-4177-b1c0-8298add53086',
              node_id=owned_node['id'])
          db_utils.create_test_volume_connector(
              uuid='65ea0296-219b-4635-b0c8-a6e055da878d',
              node_id=owned_node['id'],
              connector_id='iqn.2012-06.org.openstack.magic')
          fake_owner_allocation = db_utils.create_test_allocation(
              node_id=owned_node['id'],
              owner=owner_project_id,
              resource_class="CUSTOM_TEST")
          owned_node_history = db_utils.create_test_history(
              node_id=owned_node.id)
  
          # Leased nodes
          leased_node = db_utils.create_test_node(
              uuid=lessee_node_ident,
              owner=owner_project_id,
              lessee=lessee_project_id,
              last_error='meow',
              reservation='lolcats')
          fake_db_volume_target = db_utils.create_test_volume_target(
              node_id=leased_node['id'])
          fake_db_volume_connector = db_utils.create_test_volume_connector(
              node_id=leased_node['id'])
          fake_db_port = db_utils.create_test_port(
              node_id=leased_node['id'])
          fake_db_portgroup = db_utils.create_test_portgroup(
              node_id=leased_node['id'])
          fake_trait = 'CUSTOM_MEOW'
          fake_vif_port_id = "0e21d58f-5de2-4956-85ff-33935ea1ca01"
          fake_allocation_id = 61
          fake_leased_allocation = db_utils.create_test_allocation(
              id=fake_allocation_id,
              owner=lessee_project_id,
              resource_class="CUSTOM_LEASED")
  
          dbapi = db_api.get_instance()
          dbapi.update_allocation(fake_allocation_id,
                                  dict(node_id=leased_node['id']))
  
          leased_node_history = db_utils.create_test_history(
              node_id=leased_node.id)
  
          # Random objects that shouldn't be project visible
          other_node = db_utils.create_test_node(
              uuid='573208e5-cd41-4e26-8f06-ef44022b3793')
          other_port = db_utils.create_test_port(
              node_id=other_node['id'],
              uuid='abfd8dbb-1732-449a-b760-2224035c6b99',
              address='00:00:00:00:00:ff')
          other_pgroup = db_utils.create_test_portgroup(
              uuid='5810f41c-6585-41fc-b9c9-a94f50d421b5',
              node_id=other_node['id'],
              name='corgis_rule_the_world',
              address='ff:ff:ff:ff:ff:0f')
  
          self.format_data.update({
              'node_ident': unowned_node['uuid'],
              'owner_node_ident': owner_node_ident,
              'lessee_node_ident': lessee_node_ident,
              'allocated_node_ident': lessee_node_ident,
              'volume_target_ident': fake_db_volume_target['uuid'],
              'volume_connector_ident': fake_db_volume_connector['uuid'],
              'lessee_port_ident': fake_db_port['uuid'],
              'lessee_portgroup_ident': fake_db_portgroup['uuid'],
              'trait': fake_trait,
              'vif_ident': fake_vif_port_id,
              'ind_component': 'component',
              'ind_ident': 'magic_light',
              'owner_port_ident': owned_node_port['uuid'],
              'other_port_ident': other_port['uuid'],
              'owner_portgroup_ident': owner_pgroup['uuid'],
              'other_portgroup_ident': other_pgroup['uuid'],
              'driver_name': 'fake-driverz',
              'owner_allocation': fake_owner_allocation['uuid'],
              'lessee_allocation': fake_leased_allocation['uuid'],
              'owned_history_ident': owned_node_history['uuid'],
              'lessee_history_ident': leased_node_history['uuid']})
  
      @ddt.file_data('test_rbac_project_scoped.yaml')
      @ddt.unpack
      def test_project_scoped(self, **kwargs):
          self._check_skip(**kwargs)
          self._test_request(**kwargs)
  