rwlaunchpad/ra/pytest/ns/rbac/test_rbac_mano_xpath_access.py

   1 #!/usr/bin/env python3
   2 """
   3 #
   4 #   Copyright 2017 RIFT.IO Inc
   5 #
   6 #   Licensed under the Apache License, Version 2.0 (the "License");
   7 #   you may not use this file except in compliance with the License.
   8 #   You may obtain a copy of the License at
   9 #
  10 #       http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 #   Unless required by applicable law or agreed to in writing, software
  13 #   distributed under the License is distributed on an "AS IS" BASIS,
  14 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15 #   See the License for the specific language governing permissions and
  16 #   limitations under the License.
  17 #
  18 """
  19
  20 import pytest
  21 import gi
  22
  23 import rift.auto.mano
  24 import rift.auto.descriptor
  25
  26 gi.require_version('RwProjectNsdYang', '1.0')
  27 gi.require_version('RwProjectVnfdYang', '1.0')
  28 gi.require_version('RwCloudYang', '1.0')
  29 gi.require_version('RwSdnYang', '1.0')
  30 gi.require_version('RwLaunchpadYang', '1.0')
  31 gi.require_version('RwVnfrYang', '1.0')
  32 gi.require_version('RwNsrYang', '1.0')
  33 gi.require_version('RwImageMgmtYang', '1.0')
  34 gi.require_version('RwStagingMgmtYang', '1.0')
  35 gi.require_version('RwPkgMgmtYang', '1.0')
  36
  37 from gi.repository import (
  38     RwProjectNsdYang,
  39     RwProjectVnfdYang,
  40     RwCloudYang,
  41     RwSdnYang,
  42     RwLaunchpadYang,
  43     RwVnfrYang,
  44     RwNsrYang,
  45     RwImageMgmtYang,
  46     RwStagingMgmtYang,
  47     RwPkgMgmtYang,
  48 )
  49
  50 gi.require_version('RwKeyspec', '1.0')
  51 from gi.repository.RwKeyspec import quoted_key
  52
  53
  54 @pytest.fixture(scope='module')
  55 def mano_xpaths():
  56     """All xpaths which need to be accessed by users with various roles"""
  57
  58     xpaths_dict = {
  59         'catalog' : ('/vnfd-catalog', '/nsd-catalog'),
  60         'accounts' : ('/cloud', '/sdn'),
  61         'records' : ('/vnfr-catalog', '/vnfr-console', '/ns-instance-config', '/ns-instance-opdata'),
  62         'pkg-mgmt' : ('/staging-areas', '/upload-jobs', '/copy-jobs', '/download-jobs'),
  63         'config-agent': ('/config-agent',),
  64         'ro' : ('/resource-orchestrator',),
  65         'datacenter' : ('/datacenters',),
  66     }
  67     return xpaths_dict
  68
  69
  70 @pytest.fixture(scope='module')
  71 def mano_roles_xpaths_mapping():
  72     """Mano roles and its accessible xpaths mapping"""
  73     mano_roles_xpaths_mapping_dict = {
  74         'rw-project:project-admin': ('catalog', 'accounts', 'records', 'pkg-mgmt', 'config-agent', 'ro', 'datacenter'),
  75         'rw-project:project-oper' : ('catalog', 'accounts', 'records', 'pkg-mgmt', 'config-agent', 'ro', 'datacenter'),
  76         'rw-project-mano:catalog-oper' : ('catalog', 'pkg-mgmt'),
  77         'rw-project-mano:catalog-admin' : ('catalog', 'pkg-mgmt'),
  78         'rw-project-mano:lcm-admin' : ('catalog', 'accounts', 'records', 'config-agent', 'datacenter'),
  79         'rw-project-mano:lcm-oper' : ('records',),
  80         'rw-project-mano:account-admin' : ('accounts', 'config-agent', 'ro', 'datacenter'),
  81         'rw-project-mano:account-oper' : ('accounts', 'config-agent', 'ro', 'datacenter'),
  82     }
  83     return mano_roles_xpaths_mapping_dict
  84
  85
  86 @pytest.fixture(scope='module')
  87 def xpath_module_mapping():
  88     """Mano Xpaths and its module mapping. Value also carries config or opdata type along with yang-module"""
  89     xpath_module_mapping_dict = {
  90         ('/vnfd-catalog',): (RwProjectVnfdYang, 'get_config'),
  91         ('/nsd-catalog',): (RwProjectNsdYang, 'get_config'),
  92         ('/cloud',): (RwCloudYang, 'get_config'),
  93         ('/sdn',): (RwSdnYang, 'get_config'),
  94         ('/vnfr-catalog', '/vnfr-console'): (RwVnfrYang, 'get'),
  95         ('/ns-instance-config', '/ns-instance-opdata'): (RwNsrYang, 'get'),
  96         ('/upload-jobs', '/download-jobs'): (RwImageMgmtYang, 'get'),
  97         ('/copy-jobs', ): (RwPkgMgmtYang, 'get'),
  98         ('/staging-areas',): (RwStagingMgmtYang, 'get'),
  99         ('/resource-orchestrator', '/datacenters'): (RwLaunchpadYang, None),
 100         ('/config-agent',): None,
 101     }
 102     return xpath_module_mapping_dict
 103
 104 @pytest.mark.setup('mano_xpath_access')
 105 @pytest.mark.depends('nsr')
 106 @pytest.mark.incremental
 107 class TestRbacManoXpathAccess(object):
 108     def test_copy_nsd_catalog_item(self, mgmt_session):
 109         """Copy a NSD catalog item, so that /copy-jobs xpath can be tested."""
 110         nsd_path = '/rw-project:project[rw-project:name="default"]/nsd-catalog'
 111         nsd = mgmt_session.proxy(RwProjectNsdYang).get_config(nsd_path)
 112         nsd_pkg_id = nsd.nsd[0].id
 113         rpc_input = RwPkgMgmtYang.YangInput_RwPkgMgmt_PackageCopy.from_dict(
 114             {'package_type': 'NSD', 'package_id': nsd_pkg_id, 'package_name': 'test_nsd_copy',
 115              'project_name': 'default'})
 116         mgmt_session.proxy(RwPkgMgmtYang).rpc(rpc_input)
 117
 118     def test_rbac_mano_xpaths_access(self, mano_xpaths, logger, mano_roles_xpaths_mapping, xpath_module_mapping, session_class,
 119         project_keyed_xpath, user_domain, rbac_platform_proxy, rw_project_proxy, rbac_user_passwd, confd_host, rw_user_proxy, rw_rbac_int_proxy):
 120         """Verify Mano roles/Permission mapping works (Verifies only read access for all Xpaths)."""
 121         project_name = 'default'
 122
 123         # Skipping download-jobs as it is not yet implemented from MANO side.
 124         # Others are skipped becuase they need Juju, Openmano configurations etc.
 125         skip_xpaths = ('/download-jobs', '/config-agent', '/resource-orchestrator', '/datacenters', '/upload-jobs')
 126
 127         for index, (role, xpath_keys_tuple) in enumerate(mano_roles_xpaths_mapping.items()):
 128             # Create an user and assign a role
 129             user_name = 'user-{}'.format(index)
 130             rift.auto.mano.create_user(rw_user_proxy, user_name, rbac_user_passwd, user_domain)
 131             logger.debug('Creating an user {} with role {}'.format(user_name, role))
 132             if 'platform' in role:
 133                 rift.auto.mano.assign_platform_role_to_user(rbac_platform_proxy, role, user_name, user_domain, rw_rbac_int_proxy)
 134             else:
 135                 rift.auto.mano.assign_project_role_to_user(rw_project_proxy, role, user_name, project_name, user_domain, rw_rbac_int_proxy)
 136
 137             # Get user session
 138             user_session = rift.auto.mano.get_session(session_class, confd_host, user_name, rbac_user_passwd)
 139
 140             # go through each of its xpaths keys and try to access
 141             for xpath_key in xpath_keys_tuple:
 142                 for xpath in mano_xpaths[xpath_key]:
 143                     if xpath in skip_xpaths:
 144                         continue
 145                     logger.debug('User {} with role {} trying to access xpath {}'.format(user_name, role, xpath))
 146                     yang_module, get_type = [yang_module for xpath_tuple, yang_module in xpath_module_mapping.items()
 147                                                                                             if xpath in xpath_tuple][0]
 148                     user_pxy = user_session.proxy(yang_module)
 149                     get_data_func = getattr(user_pxy, get_type)
 150                     assert get_data_func(project_keyed_xpath.format(project_name=quoted_key(project_name))+xpath)
 151
 152             # go through remaining xpaths keys which this user-role not part of and try to access; it should fail
 153             access_denied_xpath_keys_tuple = set(mano_xpaths.keys()).difference(xpath_keys_tuple)
 154             for xpath_key in access_denied_xpath_keys_tuple:
 155                 for xpath in mano_xpaths[xpath_key]:
 156                     if xpath in skip_xpaths:
 157                         continue
 158                     logger.debug('User {} with role {} trying to access xpath {}. It should get None'.format(user_name, role, xpath))
 159                     yang_module, get_type = [yang_module for xpath_tuple, yang_module in xpath_module_mapping.items()
 160                                                                                             if xpath in xpath_tuple][0]
 161                     user_pxy = user_session.proxy(yang_module)
 162                     get_data_func = getattr(user_pxy, get_type)
 163                     assert get_data_func(project_keyed_xpath.format(project_name=quoted_key(project_name))+xpath) is None