x
/
vmware-nsxlib
Code Issues Proposed changes

Add policy gayeway-policy support 

This resource is used to create edge firewall rules
In addition add some other properties needed for full gateway policy support

Change-Id: I749ede90230ac789c4b757d2e4d98ab9d635ecb9
This commit is contained in:
Adit Sarfaty 2019-01-31 14:14:12 +02:00
parent b89a0ca744
commit 4125bbb671
5 changed files with 173 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -552,6 +552,7 @@ class TestPolicyL4Service(NsxPolicyLibTestCase):
description = 'desc'
protocol = constants.TCP
dest_ports = [81, 82]
source_ports = [83, 84]
tags = [{'scope': 'a', 'tag': 'b'}]
with mock.patch.object(self.policy_api,
"create_with_parent") as api_call:
@ -560,6 +561,7 @@ class TestPolicyL4Service(NsxPolicyLibTestCase):
description=description,
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_ports,
tags=tags,
tenant=TEST_TENANT)
exp_srv_def = core_defs.ServiceDef(service_id=mock.ANY,
@ -573,6 +575,7 @@ class TestPolicyL4Service(NsxPolicyLibTestCase):
name='entry',
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_ports,
tenant=TEST_TENANT)
self.assert_called_with_defs(
api_call, [exp_srv_def, exp_entry_def])
@ -652,6 +655,7 @@ class TestPolicyL4Service(NsxPolicyLibTestCase):
description = 'new desc'
protocol = 'udp'
dest_ports = [555]
source_ports = [666]
entry_body = {'id': 'entry',
'l4_protocol': 'tcp'}
@ -666,6 +670,7 @@ class TestPolicyL4Service(NsxPolicyLibTestCase):
description=description,
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_ports,
tenant=TEST_TENANT)
service_def = core_defs.ServiceDef(service_id=srv_id,
@ -677,6 +682,7 @@ class TestPolicyL4Service(NsxPolicyLibTestCase):
entry_id=mock.ANY,
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_ports,
tenant=TEST_TENANT)
self.assert_called_with_defs(
@ -968,6 +974,10 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):
super(TestPolicyCommunicationMap, self).setUp()
self.resourceApi = self.policy_lib.comm_map
self.mapDef = core_defs.CommunicationMapDef
self.entryDef = core_defs.CommunicationMapEntryDef
self.resource_type = 'SecurityPolicy'
self.path_name = 'security-policies'
def test_create_another(self):
domain_id = '111'
@ -995,7 +1005,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
direction=direction,
logged=True,
tenant=TEST_TENANT)
map_def = core_defs.CommunicationMapDef(
map_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
@ -1003,7 +1013,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
category=constants.CATEGORY_APPLICATION,
tenant=TEST_TENANT)
entry_def = core_defs.CommunicationMapEntryDef(
entry_def = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id='entry',
@ -1045,7 +1055,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
logged=False,
tenant=TEST_TENANT)
map_def = core_defs.CommunicationMapDef(
map_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
@ -1053,7 +1063,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
category=category,
tenant=TEST_TENANT)
entry_def = core_defs.CommunicationMapEntryDef(
entry_def = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id='entry',
@ -1088,7 +1098,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
dest_groups=[dest_group],
tenant=TEST_TENANT)
expected_map_def = core_defs.CommunicationMapDef(
expected_map_def = self.mapDef(
domain_id=domain_id,
map_id=mock.ANY,
name=name,
@ -1096,7 +1106,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
category=constants.CATEGORY_APPLICATION,
tenant=TEST_TENANT)
expected_entry_def = core_defs.CommunicationMapEntryDef(
expected_entry_def = self.entryDef(
domain_id=domain_id,
map_id=mock.ANY,
entry_id=mock.ANY,
@ -1125,7 +1135,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
name, domain_id, description=description,
tenant=TEST_TENANT)
expected_map_def = core_defs.CommunicationMapDef(
expected_map_def = self.mapDef(
domain_id=domain_id,
map_id=mock.ANY,
name=name,
@ -1159,7 +1169,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
direction=nsx_constants.IN,
tenant=TEST_TENANT)
expected_entry_def = core_defs.CommunicationMapEntryDef(
expected_entry_def = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=mock.ANY,
@ -1171,6 +1181,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
source_groups=[source_group],
dest_groups=[dest_group],
direction=nsx_constants.IN,
scope=None,
logged=False,
tenant=TEST_TENANT)
@ -1194,7 +1205,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
sequence_number=1,
tenant=TEST_TENANT)
expected_entry_def = core_defs.CommunicationMapEntryDef(
expected_entry_def = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=mock.ANY,
@ -1206,6 +1217,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
service_ids=None,
source_groups=[source_group],
dest_groups=[dest_group],
scope=None,
logged=False,
tenant=TEST_TENANT)
@ -1236,7 +1248,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
logged=False,
tenant=TEST_TENANT)
expected_entry_def = core_defs.CommunicationMapEntryDef(
expected_entry_def = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=mock.ANY,
@ -1248,6 +1260,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
source_groups=[source_group],
dest_groups=[dest_group],
sequence_number=seq_num + 1,
scope=None,
logged=False,
tenant=TEST_TENANT)
@ -1291,7 +1304,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
category=category,
tenant=TEST_TENANT)
expected_def = core_defs.CommunicationMapDef(
expected_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
@ -1308,7 +1321,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
map_id = '222'
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete(domain_id, map_id, tenant=TEST_TENANT)
expected_def = core_defs.CommunicationMapDef(
expected_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
tenant=TEST_TENANT)
@ -1321,7 +1334,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
with mock.patch.object(self.policy_api, "delete") as api_call:
self.resourceApi.delete_entry(domain_id, map_id, entry_id,
tenant=TEST_TENANT)
expected_def = core_defs.CommunicationMapEntryDef(
expected_def = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id=entry_id,
@ -1335,7 +1348,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
return_value={'id': map_id}) as api_call:
result = self.resourceApi.get(domain_id, map_id,
tenant=TEST_TENANT)
expected_def = core_defs.CommunicationMapDef(
expected_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
tenant=TEST_TENANT)
@ -1351,7 +1364,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
obj = self.resourceApi.get_by_name(domain_id, name,
tenant=TEST_TENANT)
self.assertIsNotNone(obj)
expected_def = core_defs.CommunicationMapDef(
expected_def = self.mapDef(
domain_id=domain_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
@ -1361,7 +1374,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
with mock.patch.object(self.policy_api, "list",
return_value={'results': []}) as api_call:
result = self.resourceApi.list(domain_id, tenant=TEST_TENANT)
expected_def = core_defs.CommunicationMapDef(
expected_def = self.mapDef(
domain_id=domain_id,
tenant=TEST_TENANT)
self.assert_called_with_def(api_call, expected_def)
@ -1390,7 +1403,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
source_groups=[source_group],
dest_groups=[dest_group],
tenant=TEST_TENANT)
map_def = core_defs.CommunicationMapDef(
map_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
name=name,
@ -1398,7 +1411,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
category=category,
tenant=TEST_TENANT)
entry_def = core_defs.CommunicationMapEntryDef(
entry_def = self.entryDef(
domain_id=domain_id,
map_id=map_id,
entry_id='entry',
@ -1434,7 +1447,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
expected_body = {'id': map_id,
'description': None,
'category': category,
'resource_type': 'SecurityPolicy',
'resource_type': self.resource_type,
'rules': [{
'display_name': name,
'id': 'entry',
@ -1444,9 +1457,10 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
'destination_groups': dest_groups}]
}
url = '%s/domains/%s/security-policies/%s' % (TEST_TENANT,
domain_id,
map_id)
url = '%s/domains/%s/%s/%s' % (TEST_TENANT,
domain_id,
self.path_name,
map_id)
self.assert_json_call('PATCH', self.client, url, data=expected_body)
def test_update_entries_logged(self):
@ -1454,7 +1468,7 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
map_id = '222'
dummy_map = {'rules': [{'logged': False}]}
updated_map = {'rules': [{'logged': True}]}
map_def = core_defs.CommunicationMapDef(
map_def = self.mapDef(
domain_id=domain_id,
map_id=map_id,
tenant=TEST_TENANT)
@ -1480,11 +1494,22 @@ class TestPolicyCommunicationMap(NsxPolicyLibTestCase):
state = self.resourceApi.get_realized_state(
domain_id, map_id, tenant=TEST_TENANT)
self.assertEqual(constants.STATE_REALIZED, state)
path = "/%s/domains/%s/security-policies/%s" % (
TEST_TENANT, domain_id, map_id)
path = "/%s/domains/%s/%s/%s" % (
TEST_TENANT, domain_id, self.path_name, map_id)
api_get.assert_called_once_with(path)
class TestPolicyGatewayPolicy(TestPolicyCommunicationMap):
def setUp(self, *args, **kwargs):
super(TestPolicyGatewayPolicy, self).setUp()
self.resourceApi = self.policy_lib.gateway_policy
self.mapDef = core_defs.GatewayPolicyDef
self.entryDef = core_defs.GatewayPolicyRuleDef
self.resource_type = 'GatewayPolicy'
self.path_name = 'gateway-policies'
class TestPolicyEnforcementPoint(NsxPolicyLibTestCase):
def setUp(self, *args, **kwargs):

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

@ -72,8 +72,8 @@ class NsxPolicyLib(lib.NsxLibBase):
*args)
self.tier1_segment_port = (
core_resources.NsxPolicyTier1SegmentPortApi(*args))
self.comm_map = core_resources.NsxPolicyCommunicationMapApi(
*args)
self.comm_map = core_resources.NsxPolicyCommunicationMapApi(*args)
self.gateway_policy = core_resources.NsxPolicyGatewayPolicyApi(*args)
self.enforcement_point = core_resources.NsxPolicyEnforcementPointApi(
*args)
self.transport_zone = core_resources.NsxPolicyTransportZoneApi(

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

@ -45,6 +45,7 @@ CATEGORY_EMERGENCY = 'Emergency'
CATEGORY_INFRASTRUCTURE = 'Infrastructure'
CATEGORY_ENVIRONMENT = 'Environment'
CATEGORY_APPLICATION = 'Application'
CATEGORY_LOCAL_GW = 'LocalGatewayRules'
ACTIVE_STANDBY = 'ACTIVE_STANDBY'
ACTIVE_ACTIVE = 'ACTIVE_ACTIVE'

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

@ -1046,6 +1046,7 @@ class L4ServiceEntryDef(ServiceEntryDef):
self._set_attr_if_specified(body, 'protocol', 'l4_protocol')
self._set_attr_if_specified(body, 'dest_ports', 'destination_ports')
self._set_attr_if_specified(body, 'source_ports', 'source_ports')
return body
@ -1080,35 +1081,52 @@ class IPProtocolServiceEntryDef(ServiceEntryDef):
return body
class CommunicationMapDef(ResourceDef):
@property
def path_pattern(self):
return (DOMAINS_PATH_PATTERN + "%s/security-policies/")
class SecurityPolicyBaseDef(ResourceDef):
@property
def path_ids(self):
return ('tenant', 'domain_id', 'map_id')
@staticmethod
def resource_type():
return 'SecurityPolicy'
def path_defs(self):
return (TenantDef, DomainDef)
def get_obj_dict(self):
body = super(CommunicationMapDef, self).get_obj_dict()
body = super(SecurityPolicyBaseDef, self).get_obj_dict()
self._set_attr_if_specified(body, 'category')
return body
class CommunicationMapDef(SecurityPolicyBaseDef):
"""AKA security policy"""
@property
def path_pattern(self):
return (DOMAINS_PATH_PATTERN + "%s/security-policies/")
@staticmethod
def resource_type():
return 'SecurityPolicy'
@staticmethod
def sub_entries_path():
return CommunicationMapEntryDef().get_last_section_dict_key
class CommunicationMapEntryDef(ResourceDef):
class GatewayPolicyDef(SecurityPolicyBaseDef):
@property
def path_pattern(self):
return (DOMAINS_PATH_PATTERN + "%s/gateway-policies/")
@staticmethod
def resource_type():
return 'GatewayPolicy'
@staticmethod
def sub_entries_path():
return GatewayPolicyRuleDef().get_last_section_dict_key
class SecurityPolicyRuleBaseDef(ResourceDef):
def get_groups_path(self, domain_id, group_ids):
if not group_ids:
return [constants.ANY_GROUP]
@ -1129,11 +1147,6 @@ class CommunicationMapEntryDef(ResourceDef):
return [constants.ANY_SERVICE]
@property
def path_pattern(self):
return (DOMAINS_PATH_PATTERN +
"%s/security-policies/%s/rules/")
@property
def path_ids(self):
return ('tenant', 'domain_id', 'map_id', 'entry_id')
@ -1142,11 +1155,8 @@ class CommunicationMapEntryDef(ResourceDef):
def resource_type():
return 'Rule'
def path_defs(self):
return (TenantDef, DomainDef, CommunicationMapDef)
def get_obj_dict(self):
body = super(CommunicationMapEntryDef, self).get_obj_dict()
body = super(SecurityPolicyRuleBaseDef, self).get_obj_dict()
domain_id = self.get_attr('domain_id')
if self.has_attr('source_groups'):
body['source_groups'] = self.get_groups_path(
@ -1164,6 +1174,28 @@ class CommunicationMapEntryDef(ResourceDef):
return body
class CommunicationMapEntryDef(SecurityPolicyRuleBaseDef):
@property
def path_pattern(self):
return (DOMAINS_PATH_PATTERN +
"%s/security-policies/%s/rules/")
def path_defs(self):
return (TenantDef, DomainDef, CommunicationMapDef)
class GatewayPolicyRuleDef(SecurityPolicyRuleBaseDef):
@property
def path_pattern(self):
return (DOMAINS_PATH_PATTERN +
"%s/gateway-policies/%s/rules/")
def path_defs(self):
return (TenantDef, DomainDef, GatewayPolicyDef)
# Currently supports only NSXT
class EnforcementPointDef(ResourceDef):

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

@ -535,6 +535,7 @@ class NsxPolicyL4ServiceApi(NsxPolicyServiceBase):
description=IGNORE,
protocol=constants.TCP,
dest_ports=IGNORE,
source_ports=IGNORE,
tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
service_id = self._init_obj_uuid(service_id)
@ -548,6 +549,7 @@ class NsxPolicyL4ServiceApi(NsxPolicyServiceBase):
name=self.SINGLE_ENTRY_ID,
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_ports,
tenant=tenant)
self._create_or_store(service_def, entry_def)
@ -555,8 +557,8 @@ class NsxPolicyL4ServiceApi(NsxPolicyServiceBase):
def update(self, service_id,
name=IGNORE, description=IGNORE,
protocol=IGNORE, dest_ports=IGNORE, tags=IGNORE,
tenant=constants.POLICY_INFRA_TENANT):
protocol=IGNORE, dest_ports=IGNORE, source_ports=IGNORE,
tags=IGNORE, tenant=constants.POLICY_INFRA_TENANT):
parent_def = self._init_parent_def(
service_id=service_id,
@ -570,6 +572,7 @@ class NsxPolicyL4ServiceApi(NsxPolicyServiceBase):
entry_id=self.SINGLE_ENTRY_ID,
protocol=protocol,
dest_ports=dest_ports,
source_ports=source_ports,
tenant=tenant)
self.policy_api.create_with_parent(parent_def, entry_def)
@ -2310,15 +2313,7 @@ class NsxPolicyIpPoolApi(NsxPolicyResourceBase):
return
class NsxPolicyCommunicationMapApi(NsxPolicyResourceBase):
"""NSX Policy CommunicationMap (Under a Domain)."""
@property
def entry_def(self):
return core_defs.CommunicationMapEntryDef
@property
def parent_entry_def(self):
return core_defs.CommunicationMapDef
class NsxPolicySecurityPolicyBaseApi(NsxPolicyResourceBase):
def _get_last_seq_num(self, domain_id, map_id,
tenant=constants.POLICY_INFRA_TENANT):
@ -2345,6 +2340,7 @@ class NsxPolicyCommunicationMapApi(NsxPolicyResourceBase):
category=constants.CATEGORY_APPLICATION,
sequence_number=None, service_ids=IGNORE,
action=constants.ACTION_ALLOW,
scope=IGNORE,
source_groups=IGNORE, dest_groups=IGNORE,
direction=nsx_constants.IN_OUT,
logged=IGNORE, tags=IGNORE,
@ -2382,6 +2378,7 @@ class NsxPolicyCommunicationMapApi(NsxPolicyResourceBase):
dest_groups=dest_groups,
service_ids=service_ids,
action=action,
scope=scope,
direction=direction,
logged=logged,
tenant=tenant)
@ -2416,6 +2413,7 @@ class NsxPolicyCommunicationMapApi(NsxPolicyResourceBase):
description=None,
sequence_number=None, service_ids=None,
action=constants.ACTION_ALLOW,
scope=None,
source_groups=None, dest_groups=None,
direction=nsx_constants.IN_OUT, logged=False,
tenant=constants.POLICY_INFRA_TENANT):
@ -2430,6 +2428,7 @@ class NsxPolicyCommunicationMapApi(NsxPolicyResourceBase):
dest_groups=dest_groups,
service_ids=service_ids,
action=action,
scope=scope,
direction=direction,
logged=logged,
tenant=tenant)
@ -2456,6 +2455,7 @@ class NsxPolicyCommunicationMapApi(NsxPolicyResourceBase):
description=None, sequence_number=None, service_ids=None,
action=constants.ACTION_ALLOW,
source_groups=None, dest_groups=None,
scope=None,
direction=nsx_constants.IN_OUT,
logged=False,
tenant=constants.POLICY_INFRA_TENANT):
@ -2482,6 +2482,7 @@ class NsxPolicyCommunicationMapApi(NsxPolicyResourceBase):
dest_groups=dest_groups,
service_ids=service_ids,
action=action,
scope=scope,
direction=direction,
logged=logged,
tenant=tenant)
@ -2499,7 +2500,7 @@ class NsxPolicyCommunicationMapApi(NsxPolicyResourceBase):
def delete_entry(self, domain_id, map_id, entry_id,
tenant=constants.POLICY_INFRA_TENANT):
entry_def = core_defs.CommunicationMapEntryDef(
entry_def = self.entry_def(
domain_id=domain_id,
map_id=map_id,
entry_id=entry_id,
@ -2508,7 +2509,7 @@ class NsxPolicyCommunicationMapApi(NsxPolicyResourceBase):
def get(self, domain_id, map_id,
tenant=constants.POLICY_INFRA_TENANT, silent=False):
map_def = core_defs.CommunicationMapDef(
map_def = self.parent_entry_def(
domain_id=domain_id,
map_id=map_id,
tenant=tenant)
@ -2517,13 +2518,13 @@ class NsxPolicyCommunicationMapApi(NsxPolicyResourceBase):
def get_by_name(self, domain_id, name,
tenant=constants.POLICY_INFRA_TENANT):
"""Return first communication map entry matched by name"""
return super(NsxPolicyCommunicationMapApi, self).get_by_name(
return super(NsxPolicySecurityPolicyBaseApi, self).get_by_name(
name, domain_id, tenant=tenant)
def list(self, domain_id,
tenant=constants.POLICY_INFRA_TENANT):
"""List all the map entries of a specific domain."""
map_def = core_defs.CommunicationMapDef(
map_def = self.parent_entry_def(
domain_id=domain_id,
tenant=tenant)
return self._list(map_def)
@ -2569,10 +2570,30 @@ class NsxPolicyCommunicationMapApi(NsxPolicyResourceBase):
else:
self.policy_api.create_or_update(parent_def)
def update_entries(self, domain_id, map_id, entries,
tenant=constants.POLICY_INFRA_TENANT):
map_def = self.parent_entry_def(
domain_id=domain_id,
map_id=map_id,
tenant=tenant)
map_path = map_def.get_resource_path()
@utils.retry_upon_exception(
exceptions.StaleRevision,
max_attempts=self.policy_api.client.max_attempts)
def _update():
# Get the current data of communication map & its' entries
comm_map = self.policy_api.get(map_def)
comm_map['rules'] = [rule.get_obj_dict() for rule in entries]
# Update the entire map at the NSX
self.policy_api.client.update(map_path, comm_map)
_update()
def update_entries_logged(self, domain_id, map_id, logged,
tenant=constants.POLICY_INFRA_TENANT):
"""Update all communication map entries logged flags"""
map_def = core_defs.CommunicationMapDef(
map_def = self.parent_entry_def(
domain_id=domain_id,
map_id=map_id,
tenant=tenant)
@ -2596,29 +2617,51 @@ class NsxPolicyCommunicationMapApi(NsxPolicyResourceBase):
def get_realized_state(self, domain_id, map_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
realization_info=None):
map_def = core_defs.CommunicationMapDef(map_id=map_id,
domain_id=domain_id,
tenant=tenant)
map_def = self.parent_entry_def(map_id=map_id,
domain_id=domain_id,
tenant=tenant)
return self._get_realized_state(map_def, entity_type=entity_type,
realization_info=realization_info)
def get_realized_id(self, domain_id, map_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT,
realization_info=None):
map_def = core_defs.CommunicationMapDef(map_id=map_id,
domain_id=domain_id,
tenant=tenant)
map_def = self.parent_entry_def(map_id=map_id,
domain_id=domain_id,
tenant=tenant)
return self._get_realized_id(map_def, entity_type=entity_type,
realization_info=realization_info)
def get_realization_info(self, domain_id, map_id, entity_type=None,
tenant=constants.POLICY_INFRA_TENANT):
map_def = core_defs.CommunicationMapDef(map_id=map_id,
domain_id=domain_id,
tenant=tenant)
map_def = self.parent_entry_def(map_id=map_id,
domain_id=domain_id,
tenant=tenant)
return self._get_realization_info(map_def, entity_type=entity_type)
class NsxPolicyCommunicationMapApi(NsxPolicySecurityPolicyBaseApi):
"""NSX Policy CommunicationMap (Under a Domain). AKA Security"""
@property
def entry_def(self):
return core_defs.CommunicationMapEntryDef
@property
def parent_entry_def(self):
return core_defs.CommunicationMapDef
class NsxPolicyGatewayPolicyApi(NsxPolicySecurityPolicyBaseApi):
"""NSX Policy Gateway policy (Edge firewall)"""
@property
def entry_def(self):
return core_defs.GatewayPolicyRuleDef
@property
def parent_entry_def(self):
return core_defs.GatewayPolicyDef
class NsxPolicyEnforcementPointApi(NsxPolicyResourceBase):
"""NSX Policy Enforcement Point."""