x/vmware-nsxlib
Code Issues Proposed changes

Add ORBAC Support in Policy API

Browse Source 
Object-level RBAC Entries Support in Policy API. This resource
controls the CRUD permissions of specified user to specified resources.
URL: /policy/api/v1/aaa/object-permissions

Change-Id: If065da6e5c91fe16a563527ec2ec36c445c9afd1
This commit is contained in:
Enhao Cui 2021-04-06 17:01:31 -07:00
parent 287757d4cd
commit f0d39ed978
7 changed files with 218 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
vmware_nsxlib/tests/unit/v3/policy/test_resources.py
View File

@ -6969,3 +6969,89 @@ class TestPolicyTier0StaticRoute(NsxPolicyLibTestCase):
static_route_id=static_route_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
class TestNsxPolicyObjectRolePermissionGroup(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestNsxPolicyObjectRolePermissionGroup, self).setUp()
self.resourceApi = self.policy_lib.object_permission
def test_create(self):
name = 'orbac_test'
operation = 'none'
path_prefix = '/fake/path/prefix'
role_name = 'cloud_admin'
with mock.patch.object(self.policy_api,
"create_or_update") as api_call:
self.resourceApi.create_or_overwrite(
name, operation, path_prefix, role_name, tenant=TEST_TENANT)
expected_def = core_defs.ObjectRolePermissionGroupDef(
name=name,
operation=operation,
path_prefix=path_prefix,
role_name=role_name,
tenant=TEST_TENANT,
patch=True)
self.assert_called_with_def(api_call, expected_def)
def test_update(self):
name = 'orbac_test'
operation = 'none'
path_prefix = '/fake/path/prefix'
role_name = 'cloud_admin'
entry_body = {
"path_prefix": path_prefix,
"role_name": role_name,
"operation": "read",
"name": name
}
with mock.patch.object(self.policy_api,
"get",
return_value=entry_body),\
self.mock_create_update() as update_call:
self.resourceApi.update(
name, operation, path_prefix, role_name, tenant=TEST_TENANT)
expected_def = core_defs.ObjectRolePermissionGroupDef(
name=name,
operation=operation,
path_prefix=path_prefix,
role_name=role_name,
tenant=TEST_TENANT,
patch=True)
self.assert_called_with_def(update_call, expected_def)
def test_get(self):
self.skipTest("The action is not supported by this resource")
def test_list(self):
path_prefix = '/fake/path/prefix'
role_name = 'cloud_admin'
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(path_prefix=path_prefix,
role_name=role_name,
tenant=TEST_TENANT)
expected_def = core_defs.ObjectRolePermissionGroupDef(
path_prefix=path_prefix,
role_name=role_name,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
self.assertEqual([], result)
def test_delete(self):
path_prefix = '/fake/path/prefix'
role_name = 'cloud_admin'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(path_prefix=path_prefix,
role_name=role_name,
tenant=TEST_TENANT)
expected_def = core_defs.ObjectRolePermissionGroupDef(
path_prefix=path_prefix,
role_name=role_name,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)

1
vmware_nsxlib/v3/nsx_constants.py
View File

@ -192,3 +192,4 @@ FEATURE_NSX_POLICY_MDPROXY = 'NSX Policy Metadata Proxy'
FEATURE_NSX_POLICY_DHCP = 'NSX Policy DHCP'
FEATURE_NSX_POLICY_GLOBAL_CONFIG = 'NSX Policy Global Config'
FEATURE_NSX_POLICY_ADMIN_STATE = 'NSX Policy Segment admin state'
FEATURE_NSX_POLICY_ORBAC = 'NSX Policy ORBAC'

4
vmware_nsxlib/v3/policy/__init__.py
View File

@ -147,6 +147,8 @@ class NsxPolicyLib(lib.NsxLibBase):
self.load_balancer = lb_resources.NsxPolicyLoadBalancerApi(*args)
self.ipsec_vpn = ipsec_vpn_resources.NsxPolicyIpsecVpnApi(*args)
self.global_config = core_resources.NsxPolicyGlobalConfig(*args)
self.object_permission = (
core_resources.NsxPolicyObjectRolePermissionGroupApi(*args))
def get_nsxlib_passthrough(self):
return self.nsx_api
@ -173,6 +175,8 @@ class NsxPolicyLib(lib.NsxLibBase):
# Features available since 2.4
if (feature == nsx_constants.FEATURE_NSX_POLICY_NETWORKING):
return True
if (feature == nsx_constants.FEATURE_NSX_POLICY_ORBAC):
return True
if (version.LooseVersion(self.get_version()) >=
version.LooseVersion(nsx_constants.NSX_VERSION_2_5_0)):

1
vmware_nsxlib/v3/policy/constants.py
View File

@ -17,6 +17,7 @@ TCP = 'TCP'
UDP = 'UDP'
POLICY_INFRA_TENANT = 'infra'
POLICY_AAA_TENANT = 'aaa'
ACTION_ALLOW = 'ALLOW'
ACTION_DENY = 'DROP'

43
vmware_nsxlib/v3/policy/core_defs.py
View File

@ -75,6 +75,9 @@ TIER0_LOCALE_SERVICES_PATH_PATTERN = (TIER0S_PATH_PATTERN +
TIER1_LOCALE_SERVICES_PATH_PATTERN = (TIER1S_PATH_PATTERN +
"%s/locale-services/")
OBJECT_PERMISSIONS_PATH_PATTERN = (TENANTS_PATH_PATTERN +
"object-permissions")
class ResourceDef(object, metaclass=abc.ABCMeta):
def __init__(self, nsx_version=None, **kwargs):
@ -2780,3 +2783,43 @@ class Tier0RouteRedistributionRule(object):
body['route_map_path'] = self.route_map_path
return body
class ObjectRolePermissionGroupDef(ResourceDef):
@staticmethod
def resource_type():
return 'ObjectRolePermissionGroupDef'
# GET and DELETE accept query parameters in url, but PATCH url does not
# accept query parameters. path_prefix and role_name uniquely defines
# this resource on NSX
@property
def path_pattern(self):
path_prefix = self.get_attr('path_prefix')
role_name = self.get_attr('role_name')
if path_prefix and path_prefix.startswith('/'):
path_prefix = path_prefix[1:]
if not self.get_attr('patch') and (path_prefix or role_name):
url_query = utils.params_to_url_query(
path_prefix=path_prefix,
role_name=role_name
)
return OBJECT_PERMISSIONS_PATH_PATTERN + "?" + url_query
return OBJECT_PERMISSIONS_PATH_PATTERN
@property
def path_ids(self):
return ('tenant', 'dummy')
def get_obj_dict(self):
body = super(ObjectRolePermissionGroupDef, self).get_obj_dict()
self._set_attrs_if_specified(body, ['inheritance_disabled',
'operation',
'path_prefix',
'role_name',
'rule_disabled'])
obj_id = self.get_attr("orbac_id")
if obj_id:
body.update({"id": obj_id})
return body

76
vmware_nsxlib/v3/policy/core_resources.py
View File

@ -5095,7 +5095,7 @@ class NsxPolicyGlobalConfig(NsxPolicyResourceBase):
raise exceptions.ManagerError(details=err_msg)
def _set_l3_forwarding_mode(self, mode, tenant):
# Using PUT as PATCH is not supported for this API
# Using PUT as PATCH is not supported for this API.
config = self.get()
if config['l3_forwarding_mode'] != mode:
config['l3_forwarding_mode'] = mode
@ -5108,3 +5108,77 @@ class NsxPolicyGlobalConfig(NsxPolicyResourceBase):
def disable_ipv6(self, tenant=constants.POLICY_INFRA_TENANT):
return self._set_l3_forwarding_mode('IPV4_ONLY', tenant)
class NsxPolicyObjectRolePermissionGroupApi(NsxPolicyResourceBase):
@property
def entry_def(self):
return core_defs.ObjectRolePermissionGroupDef
# This will send a PATCH call: /policy/api/v1/aaa/object-permissions.
def create_or_overwrite(self, name, operation, path_prefix, role_name,
orbac_id=IGNORE,
description=IGNORE,
inheritance_disabled=IGNORE,
rule_disabled=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_AAA_TENANT):
orbac_def = self._init_def(name=name,
operation=operation,
path_prefix=path_prefix,
role_name=role_name,
orbac_id=orbac_id,
description=description,
inheritance_disabled=inheritance_disabled,
rule_disabled=rule_disabled,
tags=tags,
tenant=tenant,
patch=True)
self.policy_api.create_or_update(orbac_def)
# This will send a PATCH call: /policy/api/v1/aaa/object-permissions.
def update(self, name, operation, path_prefix, role_name,
orbac_id=IGNORE,
description=IGNORE,
inheritance_disabled=IGNORE,
rule_disabled=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_AAA_TENANT):
self._update(name=name,
operation=operation,
path_prefix=path_prefix,
role_name=role_name,
orbac_id=orbac_id,
description=description,
inheritance_disabled=inheritance_disabled,
rule_disabled=rule_disabled,
tags=tags,
tenant=tenant,
patch=True)
def get(self, path_prefix, role_name, tenant=constants.POLICY_AAA_TENANT):
err_msg = (_("This action is not supported"))
raise exceptions.ManagerError(details=err_msg)
# This will send a GET call:
# /policy/api/v1/aaa/object-permissions?path_prefix=...&role_name=...
def list(self, path_prefix=None, role_name=None,
tenant=constants.POLICY_AAA_TENANT):
orbac_def = self.entry_def(path_prefix=path_prefix,
role_name=role_name,
tenant=tenant)
return self._list(orbac_def)
# This will send a DELETE call:
# /policy/api/v1/aaa/object-permissions?path_prefix=...&role_name=...
# path_prefix and role_name must be specified in the url as they are
# the identifier for an ORBAC object on NSX. Otherwise, NSX will
# still return success but actually delete nothing.
def delete(self, path_prefix, role_name,
tenant=constants.POLICY_AAA_TENANT):
orbac_def = self.entry_def(path_prefix=path_prefix,
role_name=role_name,
tenant=tenant)
self._delete_with_retry(orbac_def)

8
vmware_nsxlib/v3/utils.py
View File

@ -694,6 +694,14 @@ def get_dhcp_opt_code(name):
return _supported_options.get(name)
def params_to_url_query(**kwargs):
queries = []
for key, val in kwargs.items():
if key not in ("", None) and val not in ("", None):
queries.append("%s=%s" % (key, val))
return "&".join(queries)
class APIRateLimiter(object):
def __init__(self, max_calls, period=1.0):
self._enabled = max_calls is not None