x/vmware-nsxlib
Code Issues Proposed changes

Merge "[MP] Support firewall_match for NAT rules"

Browse Source
This commit is contained in:
Zuul 2022-08-22 14:36:05 +00:00 committed by Gerrit Code Review
commit 26acee40ad
5 changed files with 137 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
vmware_nsxlib/tests/unit/v3/test_resources.py
View File

@ -971,7 +971,8 @@ class LogicalRouterTestCase(BaseTestResource):
test_constants.FAKE_ROUTER_UUID, router_body=fake_router)
self.assertEqual(test_constants.FAKE_ROUTER_FW_SEC_UUID, section_id)
def _test_nat_rule_create(self, nsx_version, add_bypas_arg=True,
def _test_nat_rule_create(self, nsx_version, add_bypass_arg=True,
firewall_match=None,
action='SNAT', expect_failure=False,
logging=False):
router = self.get_mocked_resource()
@ -987,9 +988,11 @@ class LogicalRouterTestCase(BaseTestResource):
'rule_priority': priority,
'logging': logging
}
if add_bypas_arg:
if add_bypass_arg and not firewall_match:
# Expect nat_pass to be sent to the backend
data['nat_pass'] = False
elif firewall_match:
data['firewall_match'] = firewall_match
# Ignoring 'bypass_firewall' with version 1.1
with mock.patch("vmware_nsxlib.v3.NsxLib.get_version",
@ -1000,6 +1003,7 @@ class LogicalRouterTestCase(BaseTestResource):
translated_network=translated_net,
rule_priority=priority,
bypass_firewall=False,
firewall_match=firewall_match,
display_name=display_name,
logging=logging)
except exceptions.InvalidInput as e:
@ -1017,11 +1021,11 @@ class LogicalRouterTestCase(BaseTestResource):
def test_nat_rule_create_v1(self):
# Ignoring 'bypass_firewall' with version 1.1
self._test_nat_rule_create('1.1.0', add_bypas_arg=False)
self._test_nat_rule_create('1.1.0', add_bypass_arg=False)
def test_nat_rule_create_with_logging(self):
# enable logging parameter in snat obj
self._test_nat_rule_create('1.1.0', add_bypas_arg=False, logging=True)
self._test_nat_rule_create('1.1.0', add_bypass_arg=False, logging=True)
def test_nat_rule_create_v2(self):
# Sending 'bypass_firewall' with version 1.1
@ -1036,11 +1040,29 @@ class LogicalRouterTestCase(BaseTestResource):
self._test_nat_rule_create('2.0.0', action='NO_DNAT',
expect_failure=True)
def test_nat_rule_create_invalid(self):
def test_nat_rule_create_invalid_action(self):
# NO_DNAT is supported from 2.2 & up
self._test_nat_rule_create('2.0.0', action='INVALID',
expect_failure=True)
def test_nat_rule_create_with_fw_match(self):
self._test_nat_rule_create(
'2.0.0', action='SNAT',
firewall_match=nsx_constants.NAT_FIREWALL_MATCH_EXTERNAL)
def test_nat_rule_create_with_fw_match_and_bypass(self):
# When both firewall_match and bypass_firewall are specified,
# we only want to send firewall_match to the backend
self._test_nat_rule_create(
'2.0.0', action='SNAT',
add_bypass_arg=True,
firewall_match=nsx_constants.NAT_FIREWALL_MATCH_INTERNAL)
def test_nat_rule_create_invalid_fw_match(self):
self._test_nat_rule_create('2.0.0', action='SNAT',
firewall_match='meh',
expect_failure=True)
def test_nat_rule_list(self):
router = self.get_mocked_resource()
router.list_nat_rules(test_constants.FAKE_ROUTER_UUID)
@ -1065,6 +1087,24 @@ class LogicalRouterTestCase(BaseTestResource):
data=jsonutils.dumps(data, sort_keys=True),
headers=self.default_headers())
def test_nat_rule_update_fw_match(self):
router = self.get_mocked_resource()
rule_id = '123'
with mock.patch.object(router.client, 'get',
return_value={'id': rule_id}):
router.update_nat_rule(
test_constants.FAKE_ROUTER_UUID, rule_id,
firewall_match=nsx_constants.NAT_FIREWALL_MATCH_INTERNAL)
data = {
'id': rule_id,
'firewall_match': nsx_constants.NAT_FIREWALL_MATCH_INTERNAL}
test_client.assert_json_call(
'put', router,
('https://1.2.3.4/api/v1/logical-routers/%s/nat/rules/%s' %
(test_constants.FAKE_ROUTER_UUID, rule_id)),
data=jsonutils.dumps(data, sort_keys=True),
headers=self.default_headers())
def test_delete_nat_rule_by_gw(self):
router = self.get_mocked_resource()
rule_id = '123'

27
vmware_nsxlib/tests/unit/v3/test_router.py
View File

@ -242,6 +242,24 @@ class TestRouter(nsxlib_testcase.NsxClientTestCase):
tags=None,
display_name=None)
def test_add_gw_snat_rule_fw_match(self):
logical_router_id = test_constants.FAKE_ROUTER_UUID
gw_ip = '1.1.1.1'
with mock.patch.object(self.nsxlib.router.nsxlib.logical_router,
'add_nat_rule') as add_api:
self.nsxlib.router.add_gw_snat_rule(
logical_router_id, gw_ip,
firewall_match=nsx_constants.NAT_FIREWALL_MATCH_EXTERNAL)
add_api.assert_called_with(
logical_router_id,
translated_network=gw_ip,
action="SNAT",
firewall_match=nsx_constants.NAT_FIREWALL_MATCH_EXTERNAL,
source_net=None,
rule_priority=nsx_router.GW_NAT_PRI,
tags=None,
display_name=None)
def test_update_router_edge_cluster(self):
logical_router_id = test_constants.FAKE_ROUTER_UUID
ec_id = test_constants.FAKE_EDGE_CLUSTER
@ -299,6 +317,15 @@ class TestRouter(nsxlib_testcase.NsxClientTestCase):
'1.1.1.1', '2.2.2.2')
self.assertEqual(add_rule.call_count, 2)
def test_add_fip_nat_rules_firewall_match(self):
with mock.patch.object(self.nsxlib.logical_router,
"add_nat_rule") as add_rule:
self.nsxlib.router.add_fip_nat_rules(
test_constants.FAKE_ROUTER_UUID,
'1.1.1.1', '2.2.2.2',
firewall_match=nsx_constants.NAT_FIREWALL_MATCH_EXTERNAL)
self.assertEqual(add_rule.call_count, 2)
def test_delete_fip_nat_rules(self):
with mock.patch.object(self.nsxlib.logical_router,
"delete_nat_rule_by_values") as del_rule:

22
vmware_nsxlib/v3/core_resources.py
View File

@ -599,15 +599,26 @@ class NsxLibLogicalRouter(utils.NsxLibApiBase):
arg_val=action,
arg_name='action')
def _validate_nat_rule_firewall_match(self, firewall_match):
if not firewall_match:
return
if firewall_match not in nsx_constants.NAT_FIREWALL_MATCH_VALUES:
raise exceptions.InvalidInput(
operation="Create/Update NAt rule",
arg_val=firewall_match,
arg_name='firewall_match')
def add_nat_rule(self, logical_router_id, action, translated_network,
source_net=None, dest_net=None,
enabled=True, rule_priority=None,
match_ports=None, match_protocol=None,
match_resource_type=None,
firewall_match=None,
bypass_firewall=True, logging=None,
tags=None,
display_name=None):
self._validate_nat_rule_action(action)
self._validate_nat_rule_firewall_match(firewall_match)
resource = 'logical-routers/%s/nat/rules' % logical_router_id
body = {'action': action,
'enabled': enabled,
@ -626,11 +637,16 @@ class NsxLibLogicalRouter(utils.NsxLibApiBase):
'l4_protocol': match_protocol or nsx_constants.TCP}
# nat_pass parameter is supported with the router firewall feature
# this parameter is deprecated, and should never be used
if (self.nsxlib and
self.nsxlib.feature_supported(
nsx_constants.FEATURE_ROUTER_FIREWALL)):
body['nat_pass'] = bypass_firewall
elif not bypass_firewall:
# only consider nat_pass if firewall_match is not specified
if not firewall_match:
body['nat_pass'] = bypass_firewall
else:
body['firewall_match'] = firewall_match
elif not bypass_firewall or firewall_match:
LOG.error("Ignoring bypass_firewall for router %s nat rule: "
"this feature is not supported.", logical_router_id)
if tags is not None:
@ -703,6 +719,8 @@ class NsxLibLogicalRouter(utils.NsxLibApiBase):
def update_nat_rule(self, logical_router_id, nat_rule_id, **kwargs):
if 'action' in kwargs:
self._validate_nat_rule_action(kwargs['action'])
if 'firewall_match' in kwargs:
self._validate_nat_rule_firewall_match(kwargs['firewall_match'])
resource = 'logical-routers/%s/nat/rules/%s' % (
logical_router_id, nat_rule_id)
return self._update_resource(resource, kwargs, retry=True)

9
vmware_nsxlib/v3/nsx_constants.py
View File

@ -132,6 +132,15 @@ TRANSPORT_TYPE_OVERLAY = 'OVERLAY'
HOST_SWITCH_MODE_ENS = 'ENS'
HOST_SWITCH_MODE_STANDARD = 'STANDARD'
# NAT firewall match
NAT_FIREWALL_MATCH_BYPASS = 'BYPASS'
NAT_FIREWALL_MATCH_EXTERNAL = 'MATCH_EXTERNAL_ADDRESS'
NAT_FIREWALL_MATCH_INTERNAL = 'MATCH_INTERNAL_ADDRESS'
NAT_FIREWALL_MATCH_VALUES = [
NAT_FIREWALL_MATCH_BYPASS,
NAT_FIREWALL_MATCH_INTERNAL,
NAT_FIREWALL_MATCH_EXTERNAL]
# Error codes returned by the backend
ERR_CODE_OBJECT_NOT_FOUND = 202
ERR_CODE_IPAM_POOL_EXHAUSTED = 5109

40
vmware_nsxlib/v3/router.py
View File

@ -176,13 +176,24 @@ class RouterLib(object):
skip_not_found=True, strict_mode=False)
def add_gw_snat_rule(self, logical_router_id, gw_ip, source_net=None,
bypass_firewall=True, tags=None, display_name=None):
bypass_firewall=True, firewall_match=None,
tags=None, display_name=None):
if not firewall_match:
return self.nsxlib.logical_router.add_nat_rule(
logical_router_id, action="SNAT",
translated_network=gw_ip,
source_net=source_net,
rule_priority=GW_NAT_PRI,
bypass_firewall=bypass_firewall,
tags=tags,
display_name=display_name)
# Ignore bypass_firewall if firewall_match is specified
return self.nsxlib.logical_router.add_nat_rule(
logical_router_id, action="SNAT",
translated_network=gw_ip,
source_net=source_net,
rule_priority=GW_NAT_PRI,
bypass_firewall=bypass_firewall,
firewall_match=firewall_match,
tags=tags,
display_name=display_name)
@ -224,13 +235,34 @@ class RouterLib(object):
def add_fip_nat_rules(self, logical_router_id, ext_ip, int_ip,
match_ports=None, bypass_firewall=True,
firewall_match=None,
tags=None, display_name=None):
if not firewall_match:
self.nsxlib.logical_router.add_nat_rule(
logical_router_id, action="SNAT",
translated_network=ext_ip,
source_net=int_ip,
rule_priority=FIP_NAT_PRI,
bypass_firewall=bypass_firewall,
tags=tags,
display_name=display_name)
self.nsxlib.logical_router.add_nat_rule(
logical_router_id, action="DNAT",
translated_network=int_ip,
dest_net=ext_ip,
rule_priority=FIP_NAT_PRI,
match_ports=match_ports,
bypass_firewall=bypass_firewall,
tags=tags,
display_name=display_name)
return
# Ignore bypass_firewall if firewall_match is specified
self.nsxlib.logical_router.add_nat_rule(
logical_router_id, action="SNAT",
translated_network=ext_ip,
source_net=int_ip,
rule_priority=FIP_NAT_PRI,
bypass_firewall=bypass_firewall,
firewall_match=firewall_match,
tags=tags,
display_name=display_name)
self.nsxlib.logical_router.add_nat_rule(
@ -239,7 +271,7 @@ class RouterLib(object):
dest_net=ext_ip,
rule_priority=FIP_NAT_PRI,
match_ports=match_ports,
bypass_firewall=bypass_firewall,
firewall_match=firewall_match,
tags=tags,
display_name=display_name)