Fix infinite loop when enhanced policy check

- Fix so that infinite loop don't occurred in _pre_enhanced_policy_check
  when targets and credentials is set to special role.
- And add check whether targets is set to special role
  in _pre_enhanced_policy_check

Closes-Bug: #2078012
Change-Id: I410f8422b0be44cf84ffe0a6d9edef0f699d10f7

doc/source/configuration/enhanced_policy.rst

@@ -121,6 +121,9 @@ attributes.
 
       As "all" is treated as a special value, the above attribute of resource
       cannot use "all" as the attribute value.
+      If the attribute of resource is set to "all" by mistake, only users with
+      the admin role or special roles can access the resource, and other users
+      cannot access it.
 
 
 Conversion rules

tacker/common/exceptions.py

@@ -129,6 +129,10 @@ class PolicyCheckError(TackerException):
     message = _("Failed to check policy %(policy)s because %(reason)s")
 
 
+class PolicyTargetCheckError(Forbidden):
+    message = _("Failed to check policy about target because %(reason)s")
+
+
 class InUse(TackerException):
     message = _("The resource is in use")
 

tacker/policy.py

@@ -77,7 +77,7 @@ def init(conf=cfg.CONF, policy_file=None, suppress_deprecation_warnings=False):
         _ENFORCER.load_rules()
 
 
-def _pre_enhanced_policy_check(target, credentials):
+def _pre_enhanced_policy_check(target, credentials, target_chk=True):
     """Preprocesses target and credentials for enhanced tacker policy.
 
     This method does the following things:
@@ -105,14 +105,28 @@ def _pre_enhanced_policy_check(target, credentials):
             'vendor': ['*', 'company_A'],
             'tenant: ['*', 'default']
         }
-    2) Convert special value `all` to the corresponding attribute value in
+
+    2) Check whether using special role(ex: `all`) in target
+    As not assume using such in the further procedure, this method return here
+    in case check is NG.
+    However, once a object with the special role set as an attribute has been
+    registered, this check will prevent access to that object.
+    Therefore, for only user who has credentials with admin role or special
+    role(ex: `Tenant_all`) this check is skipped.
+
+    3) Convert special value `all` to the corresponding attribute value in
     target.
 
     :param target: a dictionary of the attributes of the object
         being accessed.
     :param credentials: The information about the user performing the action.
+    :param target_chk: boolean
+        True: execute check of 2)
+        False: skip check of 2)
     :return tgt: The preprocessed target is returned.
     :return user_attrs: The preprocessed credentials is returned.
+    :raises tacker.common.exceptions.PolicyTargetCheckError
+        if check of 2) is NG
     """
     if not cfg.CONF.oslo_policy.enhanced_tacker_policy:
         return target, credentials
@@ -124,6 +138,7 @@ def _pre_enhanced_policy_check(target, credentials):
 
     LOG.debug(f'target: {target}')
 
+    special_roles = {'area': r'all@\w+', 'vendor': 'all', 'tenant': 'all'}
     convert_map = {
         'area_': 'area',
         'vendor_': 'vendor',
@@ -134,7 +149,7 @@ def _pre_enhanced_policy_check(target, credentials):
         'vendor': ['*'],
         'tenant': ['*']
     }
-    # Convert special roles to enhanced policy attributes in credentials.
+    # 1) Convert special roles to enhanced policy attributes in credentials.
     for role in credentials.get('roles'):
         role = role.lower()
         for prefix, key in convert_map.items():
@@ -143,9 +158,33 @@ def _pre_enhanced_policy_check(target, credentials):
                 if attr:
                     user_attrs[key].append(attr)
 
+    # 2) Check whether using special role(ex: `all`) in target,
+    if target_chk and not credentials.get('is_admin', False):
+        for k, v in tgt.items():
+            if (k in special_roles and
+                    re.match(special_roles[k], v)):
+                chk_all = False
+                # Not NG in case credentials has special role
+                for uv in user_attrs[k]:
+                    if re.match(special_roles[k], uv):
+                        # In case 'area':
+                        # OK case1 : credential is all@all and
+                        #   target is all@region_A
+                        # OK case2 : credential is all@region_A and
+                        #   target is all@region_A
+                        # NG case  : credential is all@region_A and
+                        #   target is all@region_B
+                        if k == 'area' and uv != 'all@all' and uv != v.lower():
+                            continue
+                        chk_all = True
+                if not chk_all:
+                    msg = f'`{v}` of {k} in target is special role'
+                    LOG.error(msg)
+                    raise exceptions.PolicyTargetCheckError(reason=msg)
+
     common_keys = user_attrs.keys() & tgt.keys()
 
-    # Convert special value `all` to the corresponding attribute value in
+    # 3) Convert special value `all` to the corresponding attribute value in
     # target.
     for key in common_keys:
         tgt[key] = tgt[key].lower()
@@ -153,10 +192,11 @@ def _pre_enhanced_policy_check(target, credentials):
         if tgt.get(key) == '*':
             continue
         to_remove = []
+        orig_attrs = copy.deepcopy(attrs)
         if 'area' == key:
             if not is_valid_area(tgt.get(key)):
                 continue
-            for attr in attrs:
+            for attr in orig_attrs:
                 if not is_valid_area(attr):
                     continue
                 if 'all@all' == attr:
@@ -183,7 +223,7 @@ def _pre_enhanced_policy_check(target, credentials):
                     #       'all@region_A' -> to be removed.
 
         else:
-            for attr in attrs:
+            for attr in orig_attrs:
                 if 'all' == attr:
                     # example:
                     #   target = {'vendor': 'company_A'}
@@ -217,9 +257,16 @@ def authorize(context, action, target, do_raise=True, exc=None):
     # readable way.
     if context.system_scope:
         credentials['system'] = context.system_scope
-    target, credentials = _pre_enhanced_policy_check(target, credentials)
     if not exc:
         exc = exceptions.PolicyNotAuthorized
+
+    try:
+        target, credentials = _pre_enhanced_policy_check(target, credentials)
+    except exceptions.PolicyTargetCheckError:
+        LOG.error('Policy check for %(action)s failed with target check '
+              '%(target)s', {'action': action, 'target': target})
+        raise exc(action=action)
+
     try:
         result = _ENFORCER.authorize(action, target, credentials,
                                      do_raise=do_raise, exc=exc, action=action)
@@ -554,7 +601,12 @@ def check(context, action, target, plugin=None, might_not_exist=False,
             target.update({'area': area})
     if 'tenant_id' in target:
         target['project_id'] = target['tenant_id']
-    target, credentials = _pre_enhanced_policy_check(target, credentials)
+    try:
+        target, credentials = _pre_enhanced_policy_check(target, credentials)
+    except exceptions.PolicyTargetCheckError:
+        LOG.error('Policy check for %(action)s failed with target check '
+              '%(target)s', {'action': action, 'target': target})
+        return False
 
     result = _ENFORCER.enforce(match_rule,
                                target,
@@ -602,8 +654,21 @@ def enforce(context, action, target, plugin=None, pluralized=None,
             target.update({'area': area})
     if 'tenant_id' in target:
         target['project_id'] = target['tenant_id']
-    target, credentials = _pre_enhanced_policy_check(target, credentials)
 
+    # About target_chk:
+    # Set target_chk=False, as not execute target check.
+    # The reason is that matching with the behavior of other APIs
+    # that pass through the 'authorize' method.
+    # ex:
+    #   1 openstack vim register with area set special role
+    #     (this method is passed through, for policy check)
+    #     -> target_chk=True, policy check `NG`
+    #   2 instantiate with request parameter area set special role
+    #     (`authorize` method is passed through, for policy check)
+    #     -> target_chk=True, policy check `OK`
+    #   -> There is a difference in the behavior of 1 and 2
+    target, credentials = _pre_enhanced_policy_check(
+        target, credentials, target_chk=False)
     try:
         result = _ENFORCER.enforce(rule, target, credentials, action=action,
                                    do_raise=True, exc=exc)

tacker/tests/functional/sol_enhanced_policy/base.py

@@ -766,6 +766,36 @@ class VimAPIsTest(BaseTackerTest, BaseEnhancedPolicyTest):
         # step 9 VIM-Delete, Resource Group C / User Group admin
         self._step_vim_delete('user_admin', vim_c, 204)
 
+    def _test_vim_apis_enhanced_policy_special_role(self, vim_type, local_vim):
+        # step 1 VIM-Register, Area special role / User Group A
+        vim_a_1 = self._step_vim_register(
+            'user_a', vim_type, local_vim, 'vim_a_1', 'all@region_A', 201)
+
+        # step 2 VIM-Register, Area special role / User Group A
+        vim_a_2 = self._step_vim_register(
+            'user_a', vim_type, local_vim, 'vim_a_2', 'all@all', 201)
+
+        # step 3 VIM-Show, Area special role / User Group A
+        self._step_vim_show('user_a', vim_a_1, 404)
+
+        # step 4 VIM-Show, Area special role / User Group A
+        self._step_vim_show('user_a', vim_a_2, 404)
+
+        # step 5 VIM-Show, Area special role / User Group All
+        self._step_vim_show('user_all', vim_a_1, 200)
+
+        # step 6 VIM-Show, Area special role / User Group Admin
+        self._step_vim_show('user_admin', vim_a_2, 200)
+
+        # step 7 VIM-Delete, Area special role / User Group A
+        self._step_vim_delete('user_a', vim_a_1, 404)
+
+        # step 8 VIM-Delete, Area special role / User Group All
+        self._step_vim_delete('user_all', vim_a_1, 204)
+
+        # step 9 VIM-Delete, Area special role / User Group Admin
+        self._step_vim_delete('user_admin', vim_a_2, 204)
+
 
 class VnflcmAPIsV1Base(vnflcm_base.BaseVnfLcmTest, BaseEnhancedPolicyTest):
 

tacker/tests/functional/sol_enhanced_policy/sol/test_policy_vim_apis_openstack.py

@@ -23,3 +23,7 @@ class VimAPIsOpenstackTest(VimAPIsTest):
     def test_vim_apis_vim_without_area_openstack(self):
         self._test_vim_apis_vim_without_area_attribute(
             'openstack', 'local-vim.yaml')
+
+    def test_vim_apis_vim_with_area_openstack_special_role(self):
+        self._test_vim_apis_enhanced_policy_special_role(
+            'openstack', 'local-vim.yaml')

tacker/tests/functional/sol_enhanced_policy/sol/test_policy_vnf_package_apis.py

@@ -228,12 +228,37 @@ class BaseVnfPackageAPIsTest(BaseTackerTest, BaseEnhancedPolicyTest):
         # step 29 PKG-Delete, Resource Group B / User Group all
         self._step_pkg_delete('user_all', pkg_b, 204)
 
+    def _test_vnf_package_apis_enhanced_policy_special_role(self, csar_name):
+        # step 1 PKG-Create, Resource Group A / User Group A
+        pkg_a = self._step_pkg_create('user_a')
+
+        # step 2 PKG-Upload-content, Vendor Special Role / User Group A
+        self._step_pkg_upload_content(
+            'user_a', pkg_a, csar_name, 'all', 403)
+
+        # step 3 PKG-Upload-content, Vendor Special Role / User Group all
+        self._step_pkg_upload_content(
+            'user_all', pkg_a, csar_name, 'all', 202)
+
+        # step 4 PKG-Update, Vendor Special Role / User Group A
+        self._step_pkg_update('user_a', pkg_a, 403)
+
+        # step 5 PKG-Update, Vendor Special Role / User Group all
+        self._step_pkg_update('user_all', pkg_a, 200)
+
+        # step 6 PKG-Delete, Vendor Special Role / User Group all
+        self._step_pkg_delete('user_all', pkg_a, 204)
+
 
 class VnfPackageAPIsTest(BaseVnfPackageAPIsTest):
 
     def test_vnf_package_apis_enhanced_policy_vnf(self):
         self._test_vnf_package_apis_enhanced_policy('test_enhanced_policy')
 
+    def test_vnf_package_apis_enhanced_policy_vnf_special_role(self):
+        self._test_vnf_package_apis_enhanced_policy_special_role(
+            'test_enhanced_policy')
+
 
 class CnfPackageAPIsTest(BaseVnfPackageAPIsTest):
 

tacker/tests/functional/sol_enhanced_policy/sol/test_policy_vnflcm_apis_v2.py

@@ -41,6 +41,8 @@ class VnflcmAPIsV2VNFBase(CommonVnfLcmTest, BaseEnhancedPolicyTest):
                    'TENANT_tenant_B', 'manager'],
         'user_c': ['VENDOR_company_C', 'AREA_area_C@region_C',
                    'TENANT_tenant_C', 'manager'],
+        'user_c_1': ['VENDOR_all', 'AREA_all@all',
+                     'TENANT_tenant_C', 'manager'],
         'user_all': ['VENDOR_all', 'AREA_all@all',
                      'TENANT_all', 'manager'],
         'user_admin': ['admin']
@@ -49,6 +51,7 @@ class VnflcmAPIsV2VNFBase(CommonVnfLcmTest, BaseEnhancedPolicyTest):
         'user_a': 'tenant_A',
         'user_b': 'tenant_B',
         'user_c': 'tenant_C',
+        'user_c_1': 'all',
         'user_all': 'tenant_B',
         'user_admin': 'tenant_C'
     }
@@ -119,6 +122,10 @@ class VnflcmAPIsV2VNFBase(CommonVnfLcmTest, BaseEnhancedPolicyTest):
             change_vnfpkg_from_image_to_image_path_2, image_path=image_path,
             provider='company_C')
 
+        # for Vendor Special Role
+        cls.vnf_pkg_sr, cls.vnfd_id_sr = cls.create_vnf_package(
+            basic_lcms_min_path, image_path=image_path, provider='all')
+
     @classmethod
     def tearDownClass(cls):
         cls.delete_vnf_package(cls.vnf_pkg_a)
@@ -130,6 +137,7 @@ class VnflcmAPIsV2VNFBase(CommonVnfLcmTest, BaseEnhancedPolicyTest):
         cls.delete_vnf_package(cls.vnf_pkg_c)
         cls.delete_vnf_package(cls.vnf_pkg_c_1)
         cls.delete_vnf_package(cls.vnf_pkg_c_2)
+        cls.delete_vnf_package(cls.vnf_pkg_sr)
         BaseEnhancedPolicyTest.tearDownClass()
         super(VnflcmAPIsV2VNFBase, cls).tearDownClass()
 
@@ -739,6 +747,63 @@ class VnflcmAPIsV2VNFInstantiateWithArea(VnflcmAPIsV2VNFBase):
             sub_id, inst_id_a, inst_id_b, zone_name_list, glance_image,
             flavour_vdu_dict)
 
+    def test_vnflcm_apis_v2_vnf_with_area_in_vim_conn_info_sr_area(self):
+        # test in case area is special role, including vendor is special role
+
+        glance_image = None
+        flavour_vdu_dict = None
+        zone_name_list = None
+
+        # step 1 LCM-CreateV2, Vendor Special Role / User Group A
+        self._step_lcm_create('user_a', self.vnfd_id_sr, 403)
+
+        # step 2 LCM-CreateV2, Resource Group A / User Group A
+        inst_id_a = self._step_lcm_create('user_a', self.vnfd_id_a, 201)
+
+        # step 3 LCM-InstantiateV2, Area Special Role / User Group A
+        self._step_lcm_instantiate('user_a', inst_id_a, 'tenant_A',
+                                   glance_image, flavour_vdu_dict,
+                                   zone_name_list, 202, area='all@all')
+
+        # step 4 LCM-ShowV2, Area Special Role / User Group A
+        self._step_lcm_show('user_a', inst_id_a, 403)
+
+        # step 5 LCM-ShowV2, Area Special Role / User Group Admin
+        self._step_lcm_show('user_admin', inst_id_a, 200)
+
+        # step 6 LCM-TerminateV2, Area Special Role / User Group Admin
+        self._step_lcm_terminate('user_admin', inst_id_a, 202)
+
+        # step 7 LCM-DeleteV2, Resource Group A / User Group Admin
+        self._step_lcm_delete('user_admin', inst_id_a, 204)
+
+    def test_vnflcm_apis_v2_vnf_with_area_in_vim_conn_info_sr_tenant(self):
+        # test in case tenant: special role
+
+        glance_image = None
+        flavour_vdu_dict = None
+        zone_name_list = None
+
+        # step 1 LCM-CreateV2, Resource Group C / User Group C-1
+        inst_id_c = self._step_lcm_create('user_c_1', self.vnfd_id_c, 201)
+
+        # step 2 LCM-InstantiateV2, Tenant Special Role / User Group C-1
+        self._step_lcm_instantiate('user_c_1', inst_id_c, 'all',
+                                   glance_image, flavour_vdu_dict,
+                                   zone_name_list, 202, area='area_C@region_C')
+
+        # step 3 LCM-ShowV2, Tenant Special Role / User Group C_1
+        self._step_lcm_show('user_c_1', inst_id_c, 403)
+
+        # step 4 LCM-ShowV2, Tenant Special Role / User Group Admin
+        self._step_lcm_show('user_admin', inst_id_c, 200)
+
+        # step 5 LCM-TerminateV2, Tenant Special Role / User Group Admin
+        self._step_lcm_terminate('user_admin', inst_id_c, 202)
+
+        # step 6 LCM-DeleteV2, Resource Group C / User Group Admin
+        self._step_lcm_delete('user_admin', inst_id_c, 204)
+
 
 class VnflcmAPIsV2VNFInstantiateWithAreaInRegisteredVim(VnflcmAPIsV2VNFBase):
 

tacker/tests/unit/sol_refactored/controller/test_vnflcm_v2.py

@@ -124,6 +124,8 @@ CONF = config.CONF
 
 def get_test_data_policy_instantiate():
     rules = {POLICY_NAME.format('instantiate'): "vendor:%(vendor)s"}
+    rules_admin = {POLICY_NAME.format('instantiate'):
+                   "(vendor:%(vendor)s or role:admin)"}
     test_data = [
         # 'expected_status_code': http_client.ACCEPTED
         {
@@ -138,6 +140,18 @@ def get_test_data_policy_instantiate():
             'roles': ['VENDOR_all'],
             'expected_status_code': http_client.ACCEPTED
         },
+        {
+            'vnf_instance_updates': {'vnfProvider': 'all'},
+            'rules': rules,
+            'roles': ['VENDOR_all'],
+            'expected_status_code': http_client.ACCEPTED
+        },
+        {
+            'vnf_instance_updates': {'vnfProvider': 'all'},
+            'rules': rules_admin,
+            'roles': ['admin'],
+            'expected_status_code': http_client.ACCEPTED
+        },
         # 'expected_status_code': http_client.FORBIDDEN
         {
             'vnf_instance_updates': {'vnfProvider': 'provider_A'},
@@ -151,6 +165,12 @@ def get_test_data_policy_instantiate():
             'roles': ['VENDOR_provider_B'],
             'expected_status_code': http_client.FORBIDDEN
         },
+        {
+            'vnf_instance_updates': {'vnfProvider': 'all'},
+            'rules': rules,
+            'roles': ['VENDOR_provider_A'],
+            'expected_status_code': http_client.FORBIDDEN
+        },
     ]
     return test_data
 
@@ -163,6 +183,20 @@ def get_test_data_policy_vnf_instantiated(action, success_status_code):
         access_info={"key1": 'value1', "key2": 'value2'},
         extra={'area': 'area_A@region_A'}
     )
+    vim_connection_info_area_all_region_a = objects.VimConnectionInfo(
+        id='f8c35bd0-4d67-4436-9f11-14b8a84c92aa',
+        vimId='f8c35bd0-4d67-4436-9f11-14b8a84c92aa',
+        vimType='openstack',
+        access_info={"key1": 'value1', "key2": 'value2'},
+        extra={'area': 'all@region_A'}  # special role
+    )
+    vim_connection_info_area_all_region_all = objects.VimConnectionInfo(
+        id='f8c35bd0-4d67-4436-9f11-14b8a84c92aa',
+        vimId='f8c35bd0-4d67-4436-9f11-14b8a84c92aa',
+        vimType='openstack',
+        access_info={"key1": 'value1', "key2": 'value2'},
+        extra={'area': 'all@all'}  # special role
+    )
     vim_connection_info_without_area = objects.VimConnectionInfo(
         id='f8c35bd0-4d67-4436-9f11-14b8a84c92aa',
         vimId='f8c35bd0-4d67-4436-9f11-14b8a84c92aa',
@@ -177,11 +211,24 @@ def get_test_data_policy_vnf_instantiated(action, success_status_code):
         'vnfProvider': 'provider_A',
         'vimConnectionInfo': {'vim1': vim_connection_info_without_area}
     }
+    vnf_instance_updates_area_all_region_a = {
+        'vnfProvider': 'provider_A',
+        'vimConnectionInfo': {'vim1': vim_connection_info_area_all_region_a}
+    }
+    vnf_instance_updates_area_all_region_all = {
+        'vnfProvider': 'provider_A',
+        'vimConnectionInfo': {'vim1': vim_connection_info_area_all_region_all}
+    }
     rule_area_vendor_tenant = {
         POLICY_NAME.format(action):
             "area:%(area)s and vendor:%(vendor)s and "
             "tenant:%(tenant)s"
     }
+    rule_area_vendor_tenant_admin = {
+        POLICY_NAME.format(action):
+            "area:%(area)s and vendor:%(vendor)s and "
+            "tenant:%(tenant)s or role:admin"
+    }
     rule_vendor = {
         POLICY_NAME.format(action): "vendor:%(vendor)s"
     }
@@ -227,6 +274,62 @@ def get_test_data_policy_vnf_instantiated(action, success_status_code):
             ],
             'expected_status_code': success_status_code
         },
+        {
+            'vnf_instance_updates': vnf_instance_updates_area_all_region_a,
+            'tenant': 'tenant_A',
+            'rules': rule_area_vendor_tenant,
+            'roles': [
+                'AREA_all@region_A',
+                'VENDOR_provider_A',
+                'TENANT_tenant_A'
+            ],
+            'expected_status_code': success_status_code
+        },
+        {
+            'vnf_instance_updates': vnf_instance_updates_area_all_region_a,
+            'tenant': 'tenant_A',
+            'rules': rule_area_vendor_tenant,
+            'roles': [
+                'AREA_all@all',
+                'VENDOR_provider_A',
+                'TENANT_tenant_A'
+            ],
+            'expected_status_code': success_status_code
+        },
+        {
+            'vnf_instance_updates': vnf_instance_updates_area_all_region_all,
+            'tenant': 'tenant_A',
+            'rules': rule_area_vendor_tenant,
+            'roles': [
+                'AREA_all@all',
+                'VENDOR_provider_A',
+                'TENANT_tenant_A'
+            ],
+            'expected_status_code': success_status_code
+        },
+        {
+            'vnf_instance_updates': vnf_instance_updates_area_all_region_all,
+            'tenant': 'tenant_A',
+            'rules': rule_area_vendor_tenant_admin,
+            'roles': [
+                'AREA_all@region_A',
+                'VENDOR_provider_A',
+                'TENANT_tenant_A',
+                'admin'
+            ],
+            'expected_status_code': success_status_code
+        },
+        {
+            'vnf_instance_updates': vnf_instance_updates,
+            'tenant': 'all',  # special role
+            'rules': rule_area_vendor_tenant,
+            'roles': [
+                'AREA_area_A@region_A',
+                'VENDOR_provider_A',
+                'TENANT_all'
+            ],
+            'expected_status_code': success_status_code
+        },
         # 'expected_status_code': http_client.FORBIDDEN
         {
             'vnf_instance_updates': vnf_instance_updates,
@@ -395,7 +498,40 @@ def get_test_data_policy_vnf_instantiated(action, success_status_code):
                 'VENDOR_provider_B',
             ],
             'expected_status_code': http_client.FORBIDDEN
-        }
+        },
+        {
+            'vnf_instance_updates': vnf_instance_updates_area_all_region_all,
+            'tenant': 'tenant_A',
+            'rules': rule_area_vendor_tenant,
+            'roles': [
+                'AREA_all@region_A',
+                'VENDOR_provider_A',
+                'TENANT_tenant_A'
+            ],
+            'expected_status_code': http_client.FORBIDDEN
+        },
+        {
+            'vnf_instance_updates': vnf_instance_updates_area_all_region_a,
+            'tenant': 'tenant_A',
+            'rules': rule_area_vendor_tenant,
+            'roles': [
+                'AREA_all@region_B',
+                'VENDOR_provider_A',
+                'TENANT_tenant_A'
+            ],
+            'expected_status_code': http_client.FORBIDDEN
+        },
+        {
+            'vnf_instance_updates': vnf_instance_updates,
+            'tenant': 'all',  # special role
+            'rules': rule_area_vendor_tenant,
+            'roles': [
+                'AREA_area_A@region_A',
+                'VENDOR_provider_A',
+                'TENANT_tenant_A'
+            ],
+            'expected_status_code': http_client.FORBIDDEN
+        },
     ]
     return test_data
 
@@ -416,6 +552,12 @@ def get_test_data_policy_delete():
             'roles': ['VENDOR_all'],
             'expected_status_code': http_client.NO_CONTENT
         },
+        {
+            'vnf_instance_updates': {'vnfProvider': 'all'},  # special role
+            'rules': rules,
+            'roles': ['VENDOR_all'],
+            'expected_status_code': http_client.NO_CONTENT
+        },
         # 'expected_status_code': http_client.FORBIDDEN
         {
             'vnf_instance_updates': {'vnfProvider': 'provider_A'},
@@ -429,6 +571,12 @@ def get_test_data_policy_delete():
             'roles': ['VENDOR_provider_B'],
             'expected_status_code': http_client.FORBIDDEN
         },
+        {
+            'vnf_instance_updates': {'vnfProvider': 'all'},  # special role
+            'rules': rules,
+            'roles': ['VENDOR_provider_B'],
+            'expected_status_code': http_client.FORBIDDEN
+        },
     ]
     return test_data
 
@@ -1995,6 +2143,8 @@ class TestVnflcmV2EnhancedPolicy(TestVnflcmV2):
         request = requests.Request()
         ctx = context.Context('fake', 'fake', roles=roles)
         ctx.api_version = api_version.APIVersion("2.0.0")
+        if 'admin' in roles:
+            ctx.is_admin = True
         request.context = ctx
         return request
 

tacker/tests/unit/test_policy.py

@@ -30,8 +30,11 @@ from tacker.common import exceptions
 from tacker import context
 from tacker import manager
 from tacker import policy
+from tacker.sol_refactored.common import config
 from tacker.tests import base
 
+CONF = config.CONF
+
 
 class PolicyFileTestCase(base.BaseTestCase):
     def setUp(self):
@@ -555,3 +558,39 @@ class TackerPolicyTestCase(base.BaseTestCase):
             {'extension:provider_network:set': 'rule:admin_only'},
             dict((policy, 'rule:admin_only') for policy in
                  expected_policies))
+
+
+class EnhancedPolicyTestCace(base.BaseTestCase):
+    def setUp(self):
+        super(EnhancedPolicyTestCace, self).setUp()
+        CONF.set_override(
+            'enhanced_tacker_policy', True, group='oslo_policy')
+        policy.reset()
+        self.addCleanup(policy.reset)
+        policy.init()
+        rules = {
+            'manager_and_owner': 'role:manager and project_id:%(project_id)s',
+            'vim_attrs_cmp': 'area:%(area)s',
+            'create_vim': 'rule:manager_and_owner or role:admin',
+            'get_vim': 'rule:vim_attrs_cmp or role:admin'
+        }
+        policy.set_rules(common_policy.Rules.from_dict(rules), overwrite=True)
+        self.context = context.Context('fake', 'fake', roles=['manager'])
+        self.target = {'project_id': 'fake', 'user_id': 'fake', 'vendor': '*',
+                       'area': '*', 'tenant': '*'}
+
+    @mock.patch.object(common_policy.Enforcer, 'enforce')
+    def test_enforce_with_target_special_role(self, mock_enforce):
+        self.target['vendor'] = 'all'
+        action = 'create_vim'
+        result = policy.enforce(self.context, action, self.target)
+        self.assertTrue(result)
+        self.assertTrue(mock_enforce.called)
+
+    @mock.patch.object(common_policy.Enforcer, 'enforce')
+    def test_check_with_target_special_role(self, mock_enforce):
+        self.target['vendor'] = 'all'
+        action = 'get_vim'
+        result = policy.check(self.context, action, self.target)
+        self.assertFalse(result)
+        self.assertFalse(mock_enforce.called)