x
/
vmware-nsx
Code Issues Proposed changes

NSX|P FWaaS V2 support 

Adding FWaaS support for the Policy plugin, implementing hte NSX gateway policy

Depends-on: I97bcbd99fcced02592a6e5f10d0d43a3e99efbe6
Change-Id: I486a6f4ab766233942008b5677722fb14b8553d7
This commit is contained in:
Adit Sarfaty 2019-01-31 15:51:08 +02:00
parent 586f4d0f1d
commit d55e6c3503
15 changed files with 1273 additions and 317 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
doc/source/devstack.rst
View File

@ -58,8 +58,6 @@ Add neutron-fwaas repo as an external repository and configure following flags i
[service_providers]
service_provider = FIREWALL_V2:fwaas_db:neutron_fwaas.services.firewall.service_drivers.agents.agents.FirewallAgentDriver:default
Note - if devstack fails due to ml2_conf.ini being missing, please copy neutron/plugins/ml2/ml2_conf.ini.sample to /etc/neutron/plugins/ml2/ml2_conf.ini and stack again.
L2GW Driver
~~~~~~~~~~~
@ -219,8 +217,6 @@ Add neutron-fwaas repo as an external repository and configure following flags i
[service_providers]
service_provider = FIREWALL_V2:fwaas_db:neutron_fwaas.services.firewall.service_drivers.agents.agents.FirewallAgentDriver:default
Note - if devstack fails due to ml2_conf.ini being missing, please copy neutron/plugins/ml2/ml2_conf.ini.sample to /etc/neutron/plugins/ml2/ml2_conf.ini and stack again.
LBaaS v2 Driver
~~~~~~~~~~~~~~~
@ -298,6 +294,23 @@ Optional: Update the nsx qos_peak_bw_multiplier in nsx.ini (default value is 2.0
[NSX]
qos_peak_bw_multiplier = <i.e 10.0>
FWaaS (V2) Driver
~~~~~~~~~~~~~~~~~
Add neutron-fwaas repo as an external repository and configure following flags in ``local.conf``::
[[local|localrc]]
enable_service q-fwaas-v2
Q_SERVICE_PLUGIN_CLASSES+=,firewall_v2
[[post-config|$NEUTRON_CONF]]
[fwaas]
enabled = True
driver = vmware_nsxp_edge_v2
[service_providers]
service_provider = FIREWALL_V2:fwaas_db:neutron_fwaas.services.firewall.service_drivers.agents.agents.FirewallAgentDriver:default
NSX-TVD
-------

1
setup.cfg
View File

@ -36,6 +36,7 @@ neutron.core_plugins =
vmware_dvs = vmware_nsx.plugin:NsxDvsPlugin
vmware_nsxtvd = vmware_nsx.plugin:NsxTVDPlugin
firewall_drivers =
vmware_nsxp_edge_v2 = vmware_nsx.services.fwaas.nsx_p.edge_fwaas_driver_v2:EdgeFwaasPDriverV2
vmware_nsxv_edge_v2 = vmware_nsx.services.fwaas.nsx_v.edge_fwaas_driver_v2:EdgeFwaasVDriverV2
vmware_nsxv3_edge_v2 = vmware_nsx.services.fwaas.nsx_v3.edge_fwaas_driver_v2:EdgeFwaasV3DriverV2
vmware_nsxtvd_edge_v2 = vmware_nsx.services.fwaas.nsx_tv.edge_fwaas_driver_v2:EdgeFwaasTVDriverV2

33
vmware_nsx/plugins/common_v3/plugin.py
View File

@ -25,6 +25,7 @@ import webob.exc
from six import moves
from six import string_types
from neutron.db import agents_db
from neutron.db import agentschedulers_db
from neutron.db import allowedaddresspairs_db as addr_pair_db
from neutron.db.availability_zone import router as router_az_db
@ -42,6 +43,7 @@ from neutron.db import portsecurity_db
from neutron.db import securitygroups_db
from neutron.db import vlantransparent_db
from neutron.extensions import securitygroup as ext_sg
from neutron_lib.agent import topics
from neutron_lib.api.definitions import allowedaddresspairs as addr_apidef
from neutron_lib.api.definitions import availability_zone as az_def
from neutron_lib.api.definitions import external_net as extnet_apidef
@ -60,6 +62,7 @@ from neutron_lib.exceptions import allowedaddresspairs as addr_exc
from neutron_lib.exceptions import l3 as l3_exc
from neutron_lib.exceptions import port_security as psec_exc
from neutron_lib.plugins import utils as plugin_utils
from neutron_lib import rpc as n_rpc
from neutron_lib.services.qos import constants as qos_consts
from neutron_lib.utils import helpers
from neutron_lib.utils import net as nl_net_utils
@ -122,6 +125,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
self._network_vlans = plugin_utils.parse_network_vlan_ranges(
self._get_conf_attr('network_vlan_ranges'))
self._native_dhcp_enabled = False
self.start_rpc_listeners_called = False
def _init_native_dhcp(self):
if not self.nsxlib:
@ -174,6 +178,26 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
plugin_cfg = getattr(cfg.CONF, self.cfg_group)
return getattr(plugin_cfg, attr)
def _setup_rpc(self):
"""Should be implemented by each plugin"""
pass
def start_rpc_listeners(self):
if self.start_rpc_listeners_called:
# If called more than once - we should not create it again
return self.conn.consume_in_threads()
self._setup_rpc()
self.topic = topics.PLUGIN
self.conn = n_rpc.Connection()
self.conn.create_consumer(self.topic, self.endpoints, fanout=False)
self.conn.create_consumer(topics.REPORTS,
[agents_db.AgentExtRpcCallback()],
fanout=False)
self.start_rpc_listeners_called = True
return self.conn.consume_in_threads()
def _get_interface_network(self, context, interface_info):
is_port, is_sub = self._validate_interface_info(interface_info)
if is_port:
@ -2422,3 +2446,12 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
else:
# attach to multiple routers
raise l3_exc.RouterInterfaceAttachmentConflict(reason=err_msg)
def _router_has_edge_fw_rules(self, context, router):
if not router.gw_port_id:
# No GW -> No rule on the edge firewall
return False
if self.fwaas_callbacks and self.fwaas_callbacks.fwaas_enabled:
ports = self._get_router_interfaces(context, router.id)
return self.fwaas_callbacks.router_with_fwg(context, ports)

94
vmware_nsx/plugins/nsx_p/plugin.py
View File

@ -23,6 +23,7 @@ from oslo_log import log
from oslo_utils import excutils
from oslo_utils import uuidutils
from neutron.db import agents_db
from neutron.db import l3_db
from neutron.db.models import l3 as l3_db_models
from neutron.db.models import securitygroup as securitygroup_model # noqa
@ -75,6 +76,8 @@ from vmware_nsx.extensions import securitygrouplogging as sg_logging
from vmware_nsx.plugins.common_v3 import plugin as nsx_plugin_common
from vmware_nsx.plugins.nsx_p import availability_zones as nsxp_az
from vmware_nsx.plugins.nsx_v3 import utils as v3_utils
from vmware_nsx.services.fwaas.common import utils as fwaas_utils
from vmware_nsx.services.fwaas.nsx_p import fwaas_callbacks_v2
from vmware_nsx.services.qos.common import utils as qos_com_utils
from vmware_nsx.services.qos.nsx_v3 import driver as qos_driver
from vmware_nsx.services.qos.nsx_v3 import pol_utils as qos_utils
@ -201,6 +204,10 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
# Init QoS
qos_driver.register(qos_utils.PolicyQosNotificationsHandler())
registry.subscribe(self.spawn_complete,
resources.PROCESS,
events.AFTER_SPAWN)
# subscribe the init complete method last, so it will be called only
# if init was successful
registry.subscribe(self.init_complete,
@ -323,6 +330,20 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
def is_tvd_plugin():
return False
def _init_fwaas(self, with_rpc):
if self.fwaas_callbacks:
# already initialized
return
if fwaas_utils.is_fwaas_v2_plugin_enabled():
LOG.info("NSXp FWaaS v2 plugin enabled")
self.fwaas_callbacks = fwaas_callbacks_v2.NsxpFwaasCallbacksV2(
with_rpc)
def spawn_complete(self, resource, event, trigger, payload=None):
# Init the FWaaS support with RPC listeners for the original process
self._init_fwaas(with_rpc=True)
def init_complete(self, resource, event, trigger, payload=None):
with locking.LockManager.get_lock('plugin-init-complete'):
if self.init_is_complete:
@ -337,8 +358,16 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
if self.nsxlib:
self.nsxlib.reinitialize_cluster(resource, event, trigger,
payload=payload)
# Init the FWaaS support without RPC listeners
# for the spawn workers
self._init_fwaas(with_rpc=False)
self.init_is_complete = True
def _setup_rpc(self):
self.endpoints = [agents_db.AgentExtRpcCallback()]
def _create_network_on_backend(self, context, net_data,
transparent_vlan,
provider_data):
@ -1134,12 +1163,26 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
return self.nsxpolicy.tier0.get_edge_cluster_path(
tier0_uuid)
def service_router_has_services(self, context, router_id, router=None):
"""Check if the neutron router has any services
which require a backend service router
currently those are: SNAT, Loadbalancer, Edge firewall
"""
if not router:
router = self._get_router(context, router_id)
snat_exist = router.enable_snat
# TODO(asarfaty) - add lbaas/octavia support here
lb_exist = False
fw_exist = self._router_has_edge_fw_rules(context, router)
return snat_exist or lb_exist or fw_exist
def verify_sr_at_backend(self, router_id):
"""Check if the backend Tier1 has a service router or not"""
if self.nsxpolicy.tier1.get_edge_cluster_path(router_id):
return True
def create_service_router(self, context, router_id, router=None):
def create_service_router(self, context, router_id, router=None,
update_firewall=True):
"""Create a service router and enable standby relocation"""
if not router:
router = self._get_router(context, router_id)
@ -1166,7 +1209,12 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
LOG.warning("Failed to enable standby relocation for router "
"%s: %s", router['id'], ex)
def delete_service_router(self, router_id):
# update firewall rules (there might be FW group waiting for a
# service router)
if update_firewall:
self.update_router_firewall(context, router_id)
def delete_service_router(self, project_id, router_id):
if cfg.CONF.nsx_p.allow_passthrough:
try:
# Enable standby relocation on this router
@ -1177,6 +1225,9 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
"%s: %s", router_id, ex)
# remove the edge firewall
if self.fwaas_callbacks and self.fwaas_callbacks.fwaas_enabled:
self.fwaas_callbacks.delete_router_gateway_policy(
project_id, router_id)
self.nsxpolicy.tier1.update(router_id, disable_firewall=True)
# remove the edge cluster from the tier1 router
@ -1210,7 +1261,7 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
context.elevated(), router_id)
sr_currently_exists = self.verify_sr_at_backend(router_id)
lb_exist = False
fw_exist = False
fw_exist = self._router_has_edge_fw_rules(context, router)
actions = self._get_update_router_gw_actions(
org_tier0_uuid, orgaddr, org_enable_snat,
new_tier0_uuid, newaddr, new_enable_snat,
@ -1256,6 +1307,7 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
for subnet in router_subnets:
self._add_subnet_snat_rule(context, router_id,
subnet, gw_address_scope, newaddr)
if actions['add_no_dnat_rules']:
for subnet in router_subnets:
self._add_subnet_no_dnat_rule(context, router_id, subnet)
@ -1266,7 +1318,7 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
subnets=actions['advertise_route_connected_flag'])
if actions['remove_service_router']:
self.delete_service_router(router_id)
self.delete_service_router(router['project_id'], router_id)
def create_router(self, context, router):
r = router['router']
@ -1487,6 +1539,9 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
subnet, gw_address_scope, gw_ip)
self._add_subnet_no_dnat_rule(context, router_id, subnet)
# update firewall rules
self.update_router_firewall(context, router_id, router_db)
except Exception as ex:
with excutils.save_and_reraise_exception():
LOG.error('Failed to create router interface for network '
@ -1534,6 +1589,9 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
self._del_subnet_snat_rule(router_id, subnet)
self._del_subnet_no_dnat_rule(router_id, subnet)
# update firewall rules
self.update_router_firewall(context, router_id, router_db)
except Exception as ex:
# do not fail the neutron action
LOG.error('Failed to remove router interface for network '
@ -2217,3 +2275,31 @@ class NsxPolicyPlugin(nsx_plugin_common.NsxPluginV3Base):
def _support_vlan_router_interfaces(self):
return True
def update_router_firewall(self, context, router_id, router_db=None,
from_fw=False):
"""Rewrite all the rules in the router edge firewall
This method should be called on FWaaS v2 updates, and on router
interfaces changes.
When FWaaS is disabled, there is no need to update the NSX router FW,
as the default rule is allow-all.
"""
if not router_db:
router_db = self._get_router(context, router_id)
if (self.fwaas_callbacks and
self.fwaas_callbacks.fwaas_enabled):
# find all the relevant ports of the router for FWaaS v2
# TODO(asarfaty): Add vm ports as well
ports = self._get_router_interfaces(context, router_id)
# let the fwaas callbacks update the router FW
return self.fwaas_callbacks.update_router_firewall(
context, router_id, router_db, ports, called_from_fw=from_fw)
def get_ip_version_service_id(self, ip_version=4):
if ip_version == 4:
return NSX_P_IPV4_SERVICE_ID
else:
return NSX_P_IPV6_SERVICE_ID

30
vmware_nsx/plugins/nsx_v3/plugin.py
View File

@ -17,7 +17,7 @@ import time
import mock
import netaddr
from neutron_lib.agent import topics
from neutron_lib.api.definitions import address_scope
from neutron_lib.api.definitions import agent as agent_apidef
from neutron_lib.api.definitions import allowedaddresspairs as addr_apidef
@ -48,7 +48,6 @@ from neutron_lib import exceptions as n_exc
from neutron_lib.exceptions import l3 as l3_exc
from neutron_lib.plugins import constants as plugin_const
from neutron_lib.plugins import directory
from neutron_lib import rpc as n_rpc
from neutron_lib.services.qos import constants as qos_consts
from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api
@ -241,8 +240,6 @@ class NsxV3Plugin(nsx_plugin_common.NsxPluginV3Base,
qos_driver.register(qos_utils.QosNotificationsHandler())
self.start_rpc_listeners_called = False
self._unsubscribe_callback_events()
if cfg.CONF.api_replay_mode:
self.supported_extension_aliases.append(api_replay.ALIAS)
@ -797,22 +794,6 @@ class NsxV3Plugin(nsx_plugin_common.NsxPluginV3Base,
dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
)
def start_rpc_listeners(self):
if self.start_rpc_listeners_called:
# If called more than once - we should not create it again
return self.conn.consume_in_threads()
self._setup_rpc()
self.topic = topics.PLUGIN
self.conn = n_rpc.Connection()
self.conn.create_consumer(self.topic, self.endpoints, fanout=False)
self.conn.create_consumer(topics.REPORTS,
[agents_db.AgentExtRpcCallback()],
fanout=False)
self.start_rpc_listeners_called = True
return self.conn.consume_in_threads()
def _get_edge_cluster(self, tier0_uuid, router):
az = self._get_router_az_obj(router)
if az and az._edge_cluster_uuid:
@ -2101,15 +2082,6 @@ class NsxV3Plugin(nsx_plugin_common.NsxPluginV3Base,
'net': sub['network_id']})
raise n_exc.InvalidInput(error_message=msg)
def _router_has_edge_fw_rules(self, context, router):
if not router.gw_port_id:
# No GW -> No rule on the edge firewall
return False
if self.fwaas_callbacks and self.fwaas_callbacks.fwaas_enabled:
ports = self._get_router_interfaces(context, router.id)
return self.fwaas_callbacks.router_with_fwg(context, ports)
def verify_sr_at_backend(self, context, router_id):
nsx_router_id = nsx_db.get_nsx_router_id(context.session,
router_id)

32
vmware_nsx/services/fwaas/common/fwaas_callbacks_v2.py
View File

@ -246,3 +246,35 @@ class NsxFwaasCallbacksV2(firewall_l3_agent_v2.L3WithFWaaS):
len(fwg.get('ports', [])) <= 1):
self.fwplugin_rpc.set_firewall_group_status(
context, fwg['id'], nl_constants.INACTIVE)
class NsxCommonv3FwaasCallbacksV2(NsxFwaasCallbacksV2):
"""NSX-V3+Policy RPC callbacks for Firewall As A Service - V2."""
def should_apply_firewall_to_router(self, context, router_id):
"""Return True if the FWaaS rules should be added to this router."""
if not super(NsxCommonv3FwaasCallbacksV2,
self).should_apply_firewall_to_router(context,
router_id):
return False
# get all the relevant router info
ctx_elevated = context.elevated()
router_data = self.core_plugin.get_router(ctx_elevated, router_id)
if not router_data:
LOG.error("Couldn't read router %s data", router_id)
return False
# Check if the FWaaS driver supports this router
if not self.internal_driver.should_apply_firewall_to_router(
router_data):
return False
return True
def router_with_fwg(self, context, router_interfaces):
for port in router_interfaces:
fwg = self.get_port_fwg(context, port['id'])
if fwg and fwg.get('status') == nl_constants.ACTIVE:
return True
return False

56
vmware_nsx/services/fwaas/common/v3_utils.py Normal file
View File

@ -0,0 +1,56 @@
# Copyright 2019 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log as logging
from neutron_lib.api.definitions import constants as fwaas_consts
from vmware_nsxlib.v3 import nsx_constants
LOG = logging.getLogger(__name__)
def translate_fw_rule_action(fwaas_action, fwaas_rule_id):
"""Translate FWaaS action to NSX action"""
if fwaas_action == fwaas_consts.FWAAS_ALLOW:
return nsx_constants.FW_ACTION_ALLOW
if fwaas_action == fwaas_consts.FWAAS_DENY:
return nsx_constants.FW_ACTION_DROP
if fwaas_action == fwaas_consts.FWAAS_REJECT:
# reject is not supported by the NSX edge firewall
LOG.warning("Reject action is not supported by the NSX backend "
"for edge firewall. Using %(action)s instead for "
"rule %(id)s",
{'action': nsx_constants.FW_ACTION_DROP,
'id': fwaas_rule_id})
return nsx_constants.FW_ACTION_DROP
# Unexpected action
LOG.error("Unsupported FWAAS action %(action)s for rule %(id)s", {
'action': fwaas_action, 'id': fwaas_rule_id})
def translate_fw_rule_protocol(fwaas_protocol):
"""Translate FWaaS L4 protocol to NSX protocol"""
if fwaas_protocol.lower() == 'tcp':
return nsx_constants.TCP
if fwaas_protocol.lower() == 'udp':
return nsx_constants.UDP
if fwaas_protocol.lower() == 'icmp':
# This will cover icmpv6 too, when adding the rule.
return nsx_constants.ICMPV4
def translate_fw_rule_ports(ports):
return [ports.replace(':', '-')]

0
vmware_nsx/services/fwaas/nsx_p/__init__.py Normal file
View File

41
vmware_nsx/services/fwaas/nsx_p/edge_fwaas_driver_v2.py Normal file
View File

@ -0,0 +1,41 @@
# Copyright 2019 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from neutron_lib.plugins import directory
from oslo_log import log as logging
from vmware_nsx.services.fwaas.nsx_v3 import edge_fwaas_driver_base \
as base_driver
LOG = logging.getLogger(__name__)
FWAAS_DRIVER_NAME = 'Fwaas V2 NSX-P driver'
class EdgeFwaasPDriverV2(base_driver.CommonEdgeFwaasV3Driver):
"""NSX-P driver for Firewall As A Service V2."""
def __init__(self):
super(EdgeFwaasPDriverV2, self).__init__(FWAAS_DRIVER_NAME)
self._core_plugin = None
@property
def core_plugin(self):
"""Get the NSX-P core plugin"""
if not self._core_plugin:
self._core_plugin = directory.get_plugin()
# make sure plugin init was completed
if not self._core_plugin.init_is_complete:
self._core_plugin.init_complete(None, None, {})
return self._core_plugin

419
vmware_nsx/services/fwaas/nsx_p/fwaas_callbacks_v2.py Normal file
View File

@ -0,0 +1,419 @@
# Copyright 2019 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import random
import time
import netaddr
from oslo_log import log as logging
from neutron_lib.exceptions import firewall_v2 as exceptions
from vmware_nsx.extensions import projectpluginmap
from vmware_nsx.services.fwaas.common import fwaas_callbacks_v2 as \
com_callbacks
from vmware_nsx.services.fwaas.common import v3_utils
from vmware_nsxlib.v3 import exceptions as nsx_lib_exc
from vmware_nsxlib.v3 import nsx_constants
from vmware_nsxlib.v3.policy import constants as policy_constants
from vmware_nsxlib.v3 import utils as nsxlib_utils
LOG = logging.getLogger(__name__)
GATEWAY_POLICY_NAME = 'Tier1 %s gateway policy'
DEFAULT_RULE_NAME = 'Default LR Layer3 Rule'
DEFAULT_RULE_ID = 'default_rule'
RULE_NAME_PREFIX = 'Fwaas-'
ROUTER_FW_TAG = 'os-router-firewall'
class NsxpFwaasCallbacksV2(com_callbacks.NsxCommonv3FwaasCallbacksV2):
"""NSX-P RPC callbacks for Firewall As A Service V2."""
def __init__(self, with_rpc):
super(NsxpFwaasCallbacksV2, self).__init__(with_rpc)
self.internal_driver = None
if self.fwaas_enabled:
self.internal_driver = self.fwaas_driver
@property
def plugin_type(self):
return projectpluginmap.NsxPlugins.NSX_P
@property
def nsxpolicy(self):
return self.core_plugin.nsxpolicy
def _get_default_backend_rule(self, domain_id, router_id):
"""Return the default allow-all rule entry
This rule enrty will be added to the end of the rules list
"""
return self.nsxpolicy.gateway_policy.build_entry(
DEFAULT_RULE_NAME, domain_id, router_id,
self._get_random_rule_id(DEFAULT_RULE_ID),
description=DEFAULT_RULE_NAME,
sequence_number=None,
action=nsx_constants.FW_ACTION_ALLOW,
scope=[self.nsxpolicy.tier1.get_path(router_id)],
source_groups=None, dest_groups=None,
direction=nsx_constants.IN_OUT)
def _translate_service(self, domain_id, router_id, rule):
"""Return the NSX Policy service id matching the FW rule service.
L4 protocol service will be created per router-id & rule-id
and the service id will reflect both, as will as the L4 protocol.
This will allow the cleanup of the service by tags when the router is
detached.
"""
ip_version = rule.get('ip_version', 4)
if rule.get('protocol'):
tags = self.nsxpolicy.build_v3_tags_payload(
rule, resource_type='os-neutron-fwrule-id',
project_name=domain_id)
tags = nsxlib_utils.add_v3_tag(tags, ROUTER_FW_TAG, router_id)
l4_protocol = v3_utils.translate_fw_rule_protocol(
rule.get('protocol'))
# The L4 protocol must be a part of the service ID to allow
# changing the protocol of a rule
srv_id = '%s-%s-%s' % (rule['protocol'], router_id, rule['id'])
srv_name = 'FW_rule_%s_%s_service' % (rule['id'], rule['protocol'])
description = '%s service for FW rule %s of Tier1 %s' % (
rule['protocol'], rule['id'], router_id)
if l4_protocol in [nsx_constants.TCP, nsx_constants.UDP]:
if rule.get('destination_port') is None:
destination_ports = []
else:
destination_ports = v3_utils.translate_fw_rule_ports(
rule['destination_port'])
if rule.get('source_port') is None:
source_ports = []
else:
source_ports = v3_utils.translate_fw_rule_ports(
rule['source_port'])
self.nsxpolicy.service.create_or_overwrite(
srv_name, service_id=srv_id,
description=description,
protocol=l4_protocol,
dest_ports=destination_ports,
source_ports=source_ports,
tags=tags)
elif l4_protocol == nsx_constants.ICMPV4:
#TODO(asarfaty): Can use predefined service for ICMP
self.nsxpolicy.icmp_service.create_or_overwrite(
srv_name, service_id=srv_id,
version=ip_version,
tags=tags)
return srv_id
def _get_random_rule_id(self, rule_id):
"""Return a rule ID with random suffix to be used on the NSX
Random sequence needs to be added to rule IDs, so that PUT command
will replace all existing rules.
Keeping the same rule id will require updating the rule revision as
well.
"""
return '%s-%s' % (rule_id, str(random.randint(1, 10000000)))
def _get_rule_ips_group_id(self, rule_id, direction):
return '%s-%s' % (direction, rule_id)
def _is_empty_cidr(self, cidr, fwaas_rule_id):
net = netaddr.IPNetwork(cidr)
if ((net.version == 4 and cidr.startswith('0.0.0.0/')) or
(net.version == 6 and str(net.ip) == "::")):
LOG.warning("Unsupported FWaaS cidr %(cidr)s for rule %(id)s",
{'cidr': cidr, 'id': fwaas_rule_id})
return True
def _validate_cidr(self, cidr, fwaas_rule_id):
error_msg = (_("Illegal FWaaS cidr %(cidr)s for rule %(id)s") %
{'cidr': cidr, 'id': fwaas_rule_id})
# Validate that this is a legal & supported ipv4 / ipv6 cidr
net = netaddr.IPNetwork(cidr)
if net.version == 4:
if net.prefixlen == 0:
LOG.error(error_msg)
raise self.driver_exception(driver=self.driver_name)
elif net.version == 6:
if net.prefixlen == 0:
LOG.error(error_msg)
raise self.driver_exception(driver=self.driver_name)
else:
LOG.error(error_msg)
raise self.driver_exception(driver=self.driver_name)
def _get_rule_cidr_group(self, domain_id, router_id, rule, is_source,
is_ingress):
field = 'source_ip_address' if is_source else 'destination_ip_address'
direction_text = 'source' if is_source else 'destination'
if (rule.get(field) and
not self._is_empty_cidr(rule[field], rule['id'])):
# Create a group for ips
group_ips = rule[field]
group_id = self._get_rule_ips_group_id(rule['id'], direction_text)
self._validate_cidr(group_ips, rule['id'])
expr = self.nsxpolicy.group.build_ip_address_expression(
[group_ips])
tags = self.nsxpolicy.build_v3_tags_payload(
rule, resource_type='os-neutron-fwrule-id',
project_name=domain_id)
tags = nsxlib_utils.add_v3_tag(tags, ROUTER_FW_TAG, router_id)
self.nsxpolicy.group.create_or_overwrite_with_conditions(
"FW_rule_%s_%s" % (rule['id'], direction_text),
domain_id, group_id=group_id,
description='%s: %s' % (direction_text, group_ips),
conditions=[expr], tags=tags)
return group_id
def _create_network_group(self, domain_id, router_id, neutron_net_id):
scope_and_tag = "%s:%s" % ('os-neutron-net-id', neutron_net_id)
tags = []
tags = nsxlib_utils.add_v3_tag(tags, ROUTER_FW_TAG, router_id)
expr = self.nsxpolicy.group.build_condition(
cond_val=scope_and_tag,
cond_key=policy_constants.CONDITION_KEY_TAG,
cond_member_type=nsx_constants.TARGET_TYPE_LOGICAL_SWITCH)
group_id = '%s-%s' % (router_id, neutron_net_id)
self.nsxpolicy.group.create_or_overwrite_with_conditions(
"Segment_%s" % neutron_net_id,
domain_id,
group_id=group_id,
description='Group for segment %s' % neutron_net_id,
conditions=[expr],
tags=tags)
return group_id
def _translate_rules(self, domain_id, router_id, segment_group,
fwaas_rules, is_ingress, logged=False):
"""Translate a list of FWaaS rules to NSX rule structure"""
translated_rules = []
for rule in fwaas_rules:
if not rule['enabled']:
# skip disabled rules
continue
# Make sure the rule has a name, and it starts with the prefix
# (backend max name length is 255)
if rule.get('name'):
rule_name = RULE_NAME_PREFIX + rule['name']
else:
rule_name = RULE_NAME_PREFIX + rule['id']
rule_name = rule_name[:255]
# Set rule ID with a random suffix
rule_id = self._get_random_rule_id(rule['id'])
action = v3_utils.translate_fw_rule_action(
rule['action'], rule['id'])
if not action:
raise exceptions.FirewallInternalDriverError(
driver=self.internal_driver.driver_name)
src_group = self._get_rule_cidr_group(
domain_id, router_id, rule, is_source=True,
is_ingress=is_ingress)
if not is_ingress and not src_group:
src_group = segment_group
dest_group = self._get_rule_cidr_group(
domain_id, router_id, rule, is_source=False,
is_ingress=is_ingress)
if is_ingress and not dest_group:
dest_group = segment_group
srv_id = self._translate_service(domain_id, router_id, rule)
direction = nsx_constants.IN if is_ingress else nsx_constants.OUT
ip_protocol = (nsx_constants.IPV4 if rule.get('ip_version', 4) == 4
else nsx_constants.IPV6)
rule_entry = self.nsxpolicy.gateway_policy.build_entry(
rule_name, domain_id, router_id, rule_id,
description=rule.get('description'),
action=action,
source_groups=[src_group] if src_group else None,
dest_groups=[dest_group] if dest_group else None,
service_ids=[srv_id] if srv_id else None,
ip_protocol=ip_protocol,
logged=logged,
scope=[self.nsxpolicy.tier1.get_path(router_id)],
direction=direction)
translated_rules.append(rule_entry)
return translated_rules
def _get_port_translated_rules(self, domain_id, router_id, neutron_net_id,
firewall_group):
"""Return the list of translated FWaaS rules per port
Add the egress/ingress rules of this port +
default drop rules in each direction for this port.
"""
net_group_id = self._create_network_group(
domain_id, router_id, neutron_net_id)
port_rules = []
# Add the firewall group ingress/egress rules only if the fw is up
if firewall_group['admin_state_up']:
port_rules.extend(self._translate_rules(
domain_id, router_id, net_group_id,
firewall_group['ingress_rule_list'], is_ingress=True))
port_rules.extend(self._translate_rules(
domain_id, router_id, net_group_id,
firewall_group['egress_rule_list'], is_ingress=False))
# Add ingress/egress block rules for this port
port_rules.extend([
self.nsxpolicy.gateway_policy.build_entry(
"Block port ingress", domain_id, router_id,
self._get_random_rule_id(
DEFAULT_RULE_ID + neutron_net_id + 'ingress'),
action=nsx_constants.FW_ACTION_DROP,
dest_groups=[net_group_id],
scope=[self.nsxpolicy.tier1.get_path(router_id)],
direction=nsx_constants.IN),
self.nsxpolicy.gateway_policy.build_entry(
"Block port egress", domain_id, router_id,
self._get_random_rule_id(
DEFAULT_RULE_ID + neutron_net_id + 'egress'),
action=nsx_constants.FW_ACTION_DROP,
scope=[self.nsxpolicy.tier1.get_path(router_id)],
source_groups=[net_group_id],
direction=nsx_constants.OUT)])
return port_rules
def _set_rules_order(self, fw_rules):
# TODO(asarfaty): Consider adding vmware-nsxlib api for this
# add sequence numbers to keep rules in order
seq_num = 0
for rule in fw_rules:
rule.attrs['sequence_number'] = seq_num
seq_num += 1
def update_router_firewall(self, context, router_id, router,
router_interfaces, called_from_fw=False):
"""Rewrite all the FWaaS v2 rules in the router edge firewall
This method should be called on FWaaS updates, and on router
interfaces changes.
The purpose of called_from_fw is to differ between fw calls and other
router calls, and if it is True - add the service router accordingly.
"""
plugin = self.core_plugin
domain_id = router['project_id']
fw_rules = []
router_with_fw = False
# Add firewall rules per port attached to a firewall group
for port in router_interfaces:
# Check if this port has a firewall
fwg = self.get_port_fwg(context, port['id'])
if fwg:
router_with_fw = True
# Add the FWaaS rules for this port:ingress/egress firewall
# rules + default ingress/egress drop rule for this port
fw_rules.extend(self._get_port_translated_rules(
domain_id, router_id, port['network_id'], fwg))
# Add a default allow-all rule to all other traffic & ports
fw_rules.append(self._get_default_backend_rule(domain_id, router_id))
self._set_rules_order(fw_rules)
# Update the backend router firewall
sr_exists_on_backend = plugin.verify_sr_at_backend(router_id)
if called_from_fw:
# FW action required
if router_with_fw:
# Firewall needed and no NSX service router: create it.
if not sr_exists_on_backend:
plugin.create_service_router(
context, router_id, update_firewall=False)
sr_exists_on_backend = True
else:
# First, check if other services exist and use the sr
router_with_services = plugin.service_router_has_services(
context, router_id, router=router)
if not router_with_services and sr_exists_on_backend:
# No other services that require service router: delete it
# This also deleted the gateway policy.
self.core_plugin.delete_service_router(
context, domain_id, router_id)
sr_exists_on_backend = False
if sr_exists_on_backend:
# update the edge firewall
self.create_router_gateway_policy(context, domain_id, router_id,
router, fw_rules)
if not router_with_fw:
# Do all the cleanup once the router has no more FW rules
self.delete_router_gateway_policy(domain_id, router_id)
self.cleanup_router_fw_resources(domain_id, router_id)
def create_router_gateway_policy(self, context, domain_id, router_id,
router, fw_rules):
"""Create/Overwrite gateway policy for a router with firewall rules"""
# Check if the gateway policy already exists
try:
self.nsxpolicy.gateway_policy.get(domain_id, map_id=router_id)
except nsx_lib_exc.ResourceNotFound:
LOG.info("Going to create gateway policy for router %s", router_id)
else:
# only update the rules of this policy
self.nsxpolicy.gateway_policy.update_entries(
domain_id, router_id, fw_rules)
return
tags = self.nsxpolicy.build_v3_tags_payload(
router, resource_type='os-neutron-router-id',
project_name=context.tenant_name)
policy_name = GATEWAY_POLICY_NAME % router_id
self.nsxpolicy.gateway_policy.create_with_entries(
policy_name, domain_id, map_id=router_id,
description=policy_name,
tags=tags,
entries=fw_rules,
category=policy_constants.CATEGORY_LOCAL_GW)
def delete_router_gateway_policy(self, domain_id, router_id):
"""Delete the gateway policy associated with a router, it it exists.
Should be called when the router is deleted / FW removed from it
"""
try:
self.nsxpolicy.gateway_policy.get(domain_id, map_id=router_id)
except nsx_lib_exc.ResourceNotFound:
return
self.nsxpolicy.gateway_policy.delete(domain_id, map_id=router_id)
# Also delete all groups & services
self.cleanup_router_fw_resources(domain_id, router_id)
def cleanup_router_fw_resources(self, domain_id, router_id):
#TODO(asarfaty): Due to platform bug, gateway policy may still be
# considered present for a short while. This is a workaround till
# issue is fixed.
time.sleep(2)
tags_to_search = [{'scope': ROUTER_FW_TAG, 'tag': router_id}]
# Delete per rule & per network groups
groups = self.nsxpolicy.search_by_tags(
tags_to_search,
self.nsxpolicy.group.entry_def.resource_type())['results']
for group in groups:
self.nsxpolicy.group.delete(domain_id, group['id'])
services = self.nsxpolicy.search_by_tags(
tags_to_search,
self.nsxpolicy.service.parent_entry_def.resource_type())['results']
for srv in services:
self.nsxpolicy.service.delete(srv['id'])

245
vmware_nsx/services/fwaas/nsx_v3/edge_fwaas_driver_base.py
View File

@ -13,74 +13,55 @@
# License for the specific language governing permissions and limitations
# under the License.
import netaddr
from neutron_lib.api.definitions import constants as fwaas_consts
from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib.plugins import directory
from neutron_lib import context as n_context
from neutron_lib.exceptions import firewall_v2 as exceptions
from oslo_log import log as logging
from vmware_nsx.extensions import projectpluginmap
from vmware_nsx.services.fwaas.common import fwaas_driver_base
from vmware_nsxlib.v3 import nsx_constants as consts
LOG = logging.getLogger(__name__)
RULE_NAME_PREFIX = 'Fwaas-'
DEFAULT_RULE_NAME = 'Default LR Layer3 Rule'
#TODO(asarfaty): this base class now serves only 1 driver and can be merged
# with it
class CommonEdgeFwaasV3Driver(fwaas_driver_base.EdgeFwaasDriverBaseV2):
"""Base class for NSX-V3 driver for Firewall As A Service - V1 & V2."""
"""Base class for NSX-V3/Policy driver for Firewall As A Service V2."""
def __init__(self, driver_exception, driver_name):
def __init__(self, driver_name):
super(CommonEdgeFwaasV3Driver, self).__init__(driver_name)
self.backend_support = True
self.driver_exception = driver_exception
registry.subscribe(
self.check_backend_version,
resources.PROCESS, events.BEFORE_SPAWN)
self.driver_exception = exceptions.FirewallInternalDriverError
self._core_plugin = None
@property
def core_plugin(self):
"""Get the NSX-V3 core plugin"""
if not self._core_plugin:
self._core_plugin = directory.get_plugin()
if self._core_plugin.is_tvd_plugin():
self._core_plugin = self._core_plugin.get_plugin_by_type(
projectpluginmap.NsxPlugins.NSX_T)
if not self._core_plugin:
# The nsx-t plugin was not initialized
return
# make sure plugin init was completed
if not self._core_plugin.init_is_complete:
self._core_plugin.init_complete(None, None, {})
return self._core_plugin
"""Get the core plugin - should be implemented by each driver"""
pass
@property
def nsxlib(self):
return self.core_plugin.nsxlib
def validate_backend_version(self):
"""Validate NSX backend supports FWaaS
Can be implemented by each driver
"""
pass
@property
def nsx_firewall(self):
return self.nsxlib.firewall_section
def _update_backend_routers(self, apply_list, fwg_id):
"""Update all the affected router on the backend"""
self.validate_backend_version()
LOG.info("Updating routers firewall for firewall group %s", fwg_id)
context = n_context.get_admin_context()
routers = set()
# the apply_list is a list of tuples: routerInfo, port-id
for router_info, port_id in apply_list:
# Skip dummy entries that were added only to avoid errors
if isinstance(router_info, str):
continue
# Skip unsupported routers
if not self.should_apply_firewall_to_router(router_info.router):
continue
routers.add(router_info.router_id)
@property
def nsx_router(self):
return self.nsxlib.logical_router
def check_backend_version(self, resource, event, trigger, payload=None):
if (self.core_plugin and
not self.nsxlib.feature_supported(consts.FEATURE_ROUTER_FIREWALL)):
# router firewall is not supported
LOG.warning("FWaaS is not supported by the NSX backend (version "
"%s): Router firewall is not supported",
self.nsxlib.get_version())
self.backend_support = False
# update each router once
for router_id in routers:
self.core_plugin.update_router_firewall(context, router_id,
from_fw=True)
def should_apply_firewall_to_router(self, router_data):
"""Return True if the firewall rules should be added the router"""
@ -88,168 +69,4 @@ class CommonEdgeFwaasV3Driver(fwaas_driver_base.EdgeFwaasDriverBaseV2):
LOG.info("Cannot apply firewall to router %s with no gateway",
router_data['id'])
return False
return True
def _translate_action(self, fwaas_action, fwaas_rule_id):
"""Translate FWaaS action to NSX action"""
if fwaas_action == fwaas_consts.FWAAS_ALLOW:
return consts.FW_ACTION_ALLOW
if fwaas_action == fwaas_consts.FWAAS_DENY:
return consts.FW_ACTION_DROP
if fwaas_action == fwaas_consts.FWAAS_REJECT:
# reject is not supported by the nsx router firewall
LOG.warning("Reject action is not supported by the NSX backend "
"for router firewall. Using %(action)s instead for "
"rule %(id)s",
{'action': consts.FW_ACTION_DROP,
'id': fwaas_rule_id})
return consts.FW_ACTION_DROP
# Unexpected action
LOG.error("Unsupported FWAAS action %(action)s for rule %(id)s", {
'action': fwaas_action, 'id': fwaas_rule_id})
raise self.driver_exception(driver=self.driver_name)
def _translate_cidr(self, cidr, fwaas_rule_id):
# Validate that this is a legal & supported ipv4 / ipv6 cidr
error_msg = (_("Unsupported FWAAS cidr %(cidr)s for rule %(id)s") % {
'cidr': cidr, 'id': fwaas_rule_id})
net = netaddr.IPNetwork(cidr)
if net.version == 4:
if cidr.startswith('0.0.0.0/'):
# Treat as ANY and just log warning
LOG.warning(error_msg)
return
if net.prefixlen == 0:
LOG.error(error_msg)
raise self.driver_exception(driver=self.driver_name)
elif net.version == 6:
if str(net.ip) == "::" or net.prefixlen == 0:
LOG.error(error_msg)
raise self.driver_exception(driver=self.driver_name)
else:
LOG.error(error_msg)
raise self.driver_exception(driver=self.driver_name)
return self.nsx_firewall.get_ip_cidr_reference(
cidr,
consts.IPV6 if net.version == 6 else consts.IPV4)
def translate_addresses_to_target(self, cidrs, plugin_type,
fwaas_rule_id=None):
translated_cidrs = []
for ip in cidrs:
res = self._translate_cidr(ip, fwaas_rule_id)
if res:
translated_cidrs.append(res)
return translated_cidrs
@staticmethod
def _translate_protocol(fwaas_protocol):
"""Translate FWaaS L4 protocol to NSX protocol"""
if fwaas_protocol.lower() == 'tcp':
return consts.TCP
if fwaas_protocol.lower() == 'udp':
return consts.UDP
if fwaas_protocol.lower() == 'icmp':
# This will cover icmpv6 too, when adding the rule.
return consts.ICMPV4
@staticmethod
def _translate_ports(ports):
return [ports.replace(':', '-')]
def _translate_services(self, fwaas_rule):
l4_protocol = self._translate_protocol(fwaas_rule['protocol'])
if l4_protocol in [consts.TCP, consts.UDP]:
source_ports = []
destination_ports = []
if fwaas_rule.get('source_port'):
source_ports = self._translate_ports(
fwaas_rule['source_port'])
if fwaas_rule.get('destination_port'):
destination_ports = self._translate_ports(
fwaas_rule['destination_port'])
return [self.nsx_firewall.get_nsservice(
consts.L4_PORT_SET_NSSERVICE,
l4_protocol=l4_protocol,
source_ports=source_ports,
destination_ports=destination_ports)]
elif l4_protocol == consts.ICMPV4:
# Add both icmp v4 & v6 services
return [
self.nsx_firewall.get_nsservice(
consts.ICMP_TYPE_NSSERVICE,
protocol=consts.ICMPV4),
self.nsx_firewall.get_nsservice(
consts.ICMP_TYPE_NSSERVICE,
protocol=consts.ICMPV6),
]
def _translate_rules(self, fwaas_rules, replace_src=None,
replace_dest=None, logged=False):
translated_rules = []
for rule in fwaas_rules:
nsx_rule = {}
if not rule['enabled']:
# skip disabled rules
continue
# Make sure the rule has a name, and it starts with the prefix
# (backend max name length is 255)
if rule.get('name'):
name = RULE_NAME_PREFIX + rule['name']
else:
name = RULE_NAME_PREFIX + rule['id']
nsx_rule['display_name'] = name[:255]
if rule.get('description'):
nsx_rule['notes'] = rule['description']
nsx_rule['action'] = self._translate_action(
rule['action'], rule['id'])
if (rule.get('destination_ip_address') and
not rule['destination_ip_address'].startswith('0.0.0.0/')):
nsx_rule['destinations'] = self.translate_addresses_to_target(
[rule['destination_ip_address']], rule['id'])
elif replace_dest:
# set this value as the destination logical switch
# (only if no dest IP)
nsx_rule['destinations'] = [{'target_type': 'LogicalSwitch',
'target_id': replace_dest}]
if (rule.get('source_ip_address') and
not rule['source_ip_address'].startswith('0.0.0.0/')):
nsx_rule['sources'] = self.translate_addresses_to_target(
[rule['source_ip_address']], rule['id'])
elif replace_src:
# set this value as the source logical switch,
# (only if no source IP)
nsx_rule['sources'] = [{'target_type': 'LogicalSwitch',
'target_id': replace_src}]
if rule.get('protocol'):
nsx_rule['services'] = self._translate_services(rule)
if logged:
nsx_rule['logged'] = logged
# Set rule direction
if replace_src:
nsx_rule['direction'] = 'OUT'
elif replace_dest:
nsx_rule['direction'] = 'IN'
translated_rules.append(nsx_rule)
return translated_rules
def validate_backend_version(self):
# prevent firewall actions if the backend does not support it
if not self.backend_support:
LOG.error("The NSX backend does not support router firewall")
raise self.driver_exception(driver=self.driver_name)
def get_default_backend_rule(self, section_id, allow_all=True):
# Add default allow all rule
old_default_rule = self.nsx_firewall.get_default_rule(
section_id)
return {
'display_name': DEFAULT_RULE_NAME,
'action': (consts.FW_ACTION_ALLOW if allow_all
else consts.FW_ACTION_DROP),
'is_default': True,
'id': old_default_rule['id'] if old_default_rule else 0}

207
vmware_nsx/services/fwaas/nsx_v3/edge_fwaas_driver_v2.py
View File

@ -13,47 +13,204 @@
# License for the specific language governing permissions and limitations
# under the License.
from neutron_lib import context as n_context
import netaddr
from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib.plugins import directory
from oslo_log import log as logging
from neutron_lib.exceptions import firewall_v2 as exceptions
from vmware_nsx.extensions import projectpluginmap
from vmware_nsx.services.fwaas.common import v3_utils
from vmware_nsx.services.fwaas.nsx_v3 import edge_fwaas_driver_base \
as base_driver
from vmware_nsxlib.v3 import nsx_constants as consts
LOG = logging.getLogger(__name__)
FWAAS_DRIVER_NAME = 'Fwaas V2 NSX-V3 driver'
RULE_NAME_PREFIX = 'Fwaas-'
DEFAULT_RULE_NAME = 'Default LR Layer3 Rule'
class EdgeFwaasV3DriverV2(base_driver.CommonEdgeFwaasV3Driver):
"""NSX-V3 driver for Firewall As A Service - V2."""
def __init__(self):
exception_cls = exceptions.FirewallInternalDriverError
super(EdgeFwaasV3DriverV2, self).__init__(exception_cls,
FWAAS_DRIVER_NAME)
super(EdgeFwaasV3DriverV2, self).__init__(FWAAS_DRIVER_NAME)
registry.subscribe(
self.check_backend_version,
resources.PROCESS, events.BEFORE_SPAWN)
def _update_backend_routers(self, apply_list, fwg_id):
"""Update all the affected router on the backend"""
self.validate_backend_version()
LOG.info("Updating routers firewall for firewall group %s", fwg_id)
context = n_context.get_admin_context()
routers = set()
# the apply_list is a list of tuples: routerInfo, port-id
for router_info, port_id in apply_list:
# Skip dummy entries that were added only to avoid errors
if isinstance(router_info, str):
continue
# Skip unsupported routers
if not self.should_apply_firewall_to_router(router_info.router):
continue
routers.add(router_info.router_id)
@property
def core_plugin(self):
"""Get the NSX-V3 core plugin"""
if not self._core_plugin:
self._core_plugin = directory.get_plugin()
if self._core_plugin.is_tvd_plugin():
self._core_plugin = self._core_plugin.get_plugin_by_type(
projectpluginmap.NsxPlugins.NSX_T)
if not self._core_plugin:
# The nsx-t plugin was not initialized
return
# make sure plugin init was completed
if not self._core_plugin.init_is_complete:
self._core_plugin.init_complete(None, None, {})
return self._core_plugin
# update each router once
for router_id in routers:
self.core_plugin.update_router_firewall(context, router_id,
from_fw=True)
@property
def nsxlib(self):
return self.core_plugin.nsxlib
@property
def nsx_firewall(self):
return self.nsxlib.firewall_section
@property
def nsx_router(self):
return self.nsxlib.logical_router
def check_backend_version(self, resource, event, trigger, payload=None):
if (self.core_plugin and
not self.nsxlib.feature_supported(consts.FEATURE_ROUTER_FIREWALL)):
# router firewall is not supported
LOG.warning("FWaaS is not supported by the NSX backend (version "
"%s): Router firewall is not supported",
self.nsxlib.get_version())
self.backend_support = False
def _translate_cidr(self, cidr, fwaas_rule_id):
# Validate that this is a legal & supported ipv4 / ipv6 cidr
error_msg = (_("Unsupported FWAAS cidr %(cidr)s for rule %(id)s") % {
'cidr': cidr, 'id': fwaas_rule_id})
net = netaddr.IPNetwork(cidr)
if net.version == 4:
if cidr.startswith('0.0.0.0/'):
# Treat as ANY and just log warning
LOG.warning(error_msg)
return
if net.prefixlen == 0:
LOG.error(error_msg)
raise self.driver_exception(driver=self.driver_name)
elif net.version == 6:
if str(net.ip) == "::" or net.prefixlen == 0:
LOG.error(error_msg)
raise self.driver_exception(driver=self.driver_name)
else:
LOG.error(error_msg)
raise self.driver_exception(driver=self.driver_name)
return self.nsx_firewall.get_ip_cidr_reference(
cidr,
consts.IPV6 if net.version == 6 else consts.IPV4)
def translate_addresses_to_target(self, cidrs, plugin_type,
fwaas_rule_id=None):
translated_cidrs = []
for ip in cidrs:
res = self._translate_cidr(ip, fwaas_rule_id)
if res:
translated_cidrs.append(res)
return translated_cidrs
def _translate_services(self, fwaas_rule):
l4_protocol = v3_utils.translate_fw_rule_protocol(
fwaas_rule['protocol'])
if l4_protocol in [consts.TCP, consts.UDP]:
source_ports = []
destination_ports = []
if fwaas_rule.get('source_port'):
source_ports = v3_utils.translate_fw_rule_ports(
fwaas_rule['source_port'])
if fwaas_rule.get('destination_port'):
destination_ports = v3_utils.translate_fw_rule_ports(
fwaas_rule['destination_port'])
return [self.nsx_firewall.get_nsservice(
consts.L4_PORT_SET_NSSERVICE,
l4_protocol=l4_protocol,
source_ports=source_ports,
destination_ports=destination_ports)]
elif l4_protocol == consts.ICMPV4:
# Add both icmp v4 & v6 services
return [
self.nsx_firewall.get_nsservice(
consts.ICMP_TYPE_NSSERVICE,
protocol=consts.ICMPV4),
self.nsx_firewall.get_nsservice(
consts.ICMP_TYPE_NSSERVICE,
protocol=consts.ICMPV6),
]
def _translate_rules(self, fwaas_rules, replace_src=None,
replace_dest=None, logged=False):
translated_rules = []
for rule in fwaas_rules:
nsx_rule = {}
if not rule['enabled']:
# skip disabled rules
continue
# Make sure the rule has a name, and it starts with the prefix
# (backend max name length is 255)
if rule.get('name'):
name = RULE_NAME_PREFIX + rule['name']
else:
name = RULE_NAME_PREFIX + rule['id']
nsx_rule['display_name'] = name[:255]
if rule.get('description'):
nsx_rule['notes'] = rule['description']
nsx_rule['action'] = v3_utils.translate_fw_rule_action(
rule['action'], rule['id'])
if not nsx_rule['action']:
raise self.driver_exception(driver=self.driver_name)
if (rule.get('destination_ip_address') and
not rule['destination_ip_address'].startswith('0.0.0.0/')):
nsx_rule['destinations'] = self.translate_addresses_to_target(
[rule['destination_ip_address']], rule['id'])
elif replace_dest:
# set this value as the destination logical switch
# (only if no dest IP)
nsx_rule['destinations'] = [{'target_type': 'LogicalSwitch',
'target_id': replace_dest}]
if (rule.get('source_ip_address') and
not rule['source_ip_address'].startswith('0.0.0.0/')):
nsx_rule['sources'] = self.translate_addresses_to_target(
[rule['source_ip_address']], rule['id'])
elif replace_src:
# set this value as the source logical switch,
# (only if no source IP)
nsx_rule['sources'] = [{'target_type': 'LogicalSwitch',
'target_id': replace_src}]
if rule.get('protocol'):
nsx_rule['services'] = self._translate_services(rule)
if logged:
nsx_rule['logged'] = logged
# Set rule direction
if replace_src:
nsx_rule['direction'] = 'OUT'
elif replace_dest:
nsx_rule['direction'] = 'IN'
translated_rules.append(nsx_rule)
return translated_rules
def validate_backend_version(self):
# prevent firewall actions if the backend does not support it
if not self.backend_support:
LOG.error("The NSX backend does not support router firewall")
raise self.driver_exception(driver=self.driver_name)
def get_default_backend_rule(self, section_id, allow_all=True):
# Add default allow all rule
old_default_rule = self.nsx_firewall.get_default_rule(
section_id)
return {
'display_name': DEFAULT_RULE_NAME,
'action': (consts.FW_ACTION_ALLOW if allow_all
else consts.FW_ACTION_DROP),
'is_default': True,
'id': old_default_rule['id'] if old_default_rule else 0}
def get_port_translated_rules(self, nsx_ls_id, firewall_group,
plugin_rules):

40
vmware_nsx/services/fwaas/nsx_v3/fwaas_callbacks_v2.py
View File

@ -15,8 +15,6 @@
from oslo_log import log as logging
from neutron_lib import constants as nl_constants
from vmware_nsx.db import db as nsx_db
from vmware_nsx.extensions import projectpluginmap
from vmware_nsx.services.fwaas.common import fwaas_callbacks_v2 as \
@ -26,7 +24,7 @@ from vmware_nsx.services.fwaas.nsx_tv import edge_fwaas_driver_v2 as tv_driver
LOG = logging.getLogger(__name__)
class Nsxv3FwaasCallbacksV2(com_callbacks.NsxFwaasCallbacksV2):
class Nsxv3FwaasCallbacksV2(com_callbacks.NsxCommonv3FwaasCallbacksV2):
"""NSX-V3 RPC callbacks for Firewall As A Service - V2."""
def __init__(self, with_rpc):
@ -43,38 +41,10 @@ class Nsxv3FwaasCallbacksV2(com_callbacks.NsxFwaasCallbacksV2):
def plugin_type(self):
return projectpluginmap.NsxPlugins.NSX_T
def should_apply_firewall_to_router(self, context, router_id):
"""Return True if the FWaaS rules should be added to this router."""
if not super(Nsxv3FwaasCallbacksV2,
self).should_apply_firewall_to_router(context,
router_id):
return False
# get all the relevant router info
ctx_elevated = context.elevated()
router_data = self.core_plugin.get_router(ctx_elevated, router_id)
if not router_data:
LOG.error("Couldn't read router %s data", router_id)
return False
# Check if the FWaaS driver supports this router
if not self.internal_driver.should_apply_firewall_to_router(
router_data):
return False
return True
def get_port_rules(self, nsx_ls_id, fwg, plugin_rules):
return self.internal_driver.get_port_translated_rules(
nsx_ls_id, fwg, plugin_rules)
def router_with_fwg(self, context, router_interfaces):
for port in router_interfaces:
fwg = self.get_port_fwg(context, port['id'])
if fwg and fwg.get('status') == nl_constants.ACTIVE:
return True
return False
def update_router_firewall(self, context, nsxlib, router_id,
router_interfaces, nsx_router_id, section_id,
from_fw=False):
@ -130,11 +100,3 @@ class Nsxv3FwaasCallbacksV2(com_callbacks.NsxFwaasCallbacksV2):
exists_on_backend = False
if exists_on_backend:
nsxlib.firewall_section.update(section_id, rules=fw_rules)
def delete_port(self, context, port_id):
# Mark the FW group as inactive if this is the last port
fwg = self.get_port_fwg(context, port_id)
if (fwg and fwg.get('status') == nl_constants.ACTIVE and
len(fwg.get('ports', [])) <= 1):
self.fwplugin_rpc.set_firewall_group_status(
context, fwg['id'], nl_constants.INACTIVE)

368
vmware_nsx/tests/unit/nsx_p/test_fwaas_v2_driver.py Normal file
View File

@ -0,0 +1,368 @@
# Copyright 2019 VMware, Inc.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import copy
import mock
from neutron_lib.api.definitions import constants as fwaas_consts
from neutron_lib.plugins import directory
from oslo_utils import uuidutils
from vmware_nsx.services.fwaas.nsx_p import edge_fwaas_driver_v2
from vmware_nsx.services.fwaas.nsx_p import fwaas_callbacks_v2
from vmware_nsx.tests.unit.nsx_p import test_plugin as test_p_plugin
from vmware_nsxlib.v3 import nsx_constants as consts
FAKE_FW_ID = 'fake_fw_uuid'
FAKE_ROUTER_ID = 'fake_rtr_uuid'
FAKE_PORT_ID = 'fake_port_uuid'
FAKE_NET_ID = 'fake_net_uuid'
GW_POLICY_PATH = ("vmware_nsxlib.v3.policy.core_resources."
"NsxPolicyGatewayPolicyApi")
class NsxpFwaasTestCase(test_p_plugin.NsxPPluginTestCaseMixin):
def setUp(self):
super(NsxpFwaasTestCase, self).setUp()
self.firewall = edge_fwaas_driver_v2.EdgeFwaasPDriverV2()
self.project_id = uuidutils.generate_uuid()
self.plugin = directory.get_plugin()
self.plugin.fwaas_callbacks = fwaas_callbacks_v2.NsxpFwaasCallbacksV2(
False)
self.plugin.fwaas_callbacks.fwaas_enabled = True
self.plugin.fwaas_callbacks.fwaas_driver = self.firewall
self.plugin.fwaas_callbacks.internal_driver = self.firewall
self.plugin.init_is_complete = True
def mock_get_random_rule_id(rid):
return rid
mock.patch.object(self.plugin.fwaas_callbacks, '_get_random_rule_id',
side_effect=mock_get_random_rule_id).start()
mock.patch.object(self.plugin.nsxpolicy, 'search_by_tags',
return_value={'results': []}).start()
def _default_rule(self, seq_num):
return self.plugin.nsxpolicy.gateway_policy.build_entry(
fwaas_callbacks_v2.DEFAULT_RULE_NAME,
self.project_id, FAKE_ROUTER_ID,
fwaas_callbacks_v2.DEFAULT_RULE_ID,
description=fwaas_callbacks_v2.DEFAULT_RULE_NAME,
action=consts.FW_ACTION_ALLOW,
scope=[self.plugin.nsxpolicy.tier1.get_path(FAKE_ROUTER_ID)],
sequence_number=seq_num,
direction=consts.IN_OUT).get_obj_dict()
def _block_interface_rules(self, seq_num):
net_group_id = '%s-%s' % (FAKE_ROUTER_ID, FAKE_NET_ID)
ingress_rule = self.plugin.nsxpolicy.gateway_policy.build_entry(
"Block port ingress",
self.project_id, FAKE_ROUTER_ID,
fwaas_callbacks_v2.DEFAULT_RULE_ID + FAKE_NET_ID + 'ingress',
action=consts.FW_ACTION_DROP,
dest_groups=[net_group_id],
scope=[self.plugin.nsxpolicy.tier1.get_path(FAKE_ROUTER_ID)],
sequence_number=seq_num,
direction=consts.IN)
egress_rule = self.plugin.nsxpolicy.gateway_policy.build_entry(
"Block port egress",
self.project_id, FAKE_ROUTER_ID,
fwaas_callbacks_v2.DEFAULT_RULE_ID + FAKE_NET_ID + 'egress',
action=consts.FW_ACTION_DROP,
source_groups=[net_group_id],
scope=[self.plugin.nsxpolicy.tier1.get_path(FAKE_ROUTER_ID)],
sequence_number=seq_num + 1,
direction=consts.OUT)
return [ingress_rule.get_obj_dict(), egress_rule.get_obj_dict()]
def _fake_rules_v4(self, is_ingress=True, cidr='10.24.4.0/24',
is_conflict=False):
rule1 = {'enabled': True,
'action': 'allow',
'ip_version': 4,
'protocol': 'tcp',
'destination_port': '80',
'id': 'fake-fw-rule1',
'description': 'first rule'}
rule2 = {'name': 'rule 2',
'enabled': True,
'action': 'reject',
'ip_version': 4,
'protocol': 'tcp',
'destination_port': '22:24',
'source_port': '1:65535',
'id': 'fake-fw-rule2'}
rule3 = {'enabled': True,
'action': 'deny',
'ip_version': 4,
'protocol': 'icmp',
'id': 'fake-fw-rule3'}
rule4 = {'enabled': True,
'action': 'deny',
'ip_version': 4,
'id': 'fake-fw-rule4'}
if is_ingress:
if not is_conflict:
rule1['source_ip_address'] = cidr
else:
rule1['destination_ip_address'] = cidr
else:
if not is_conflict:
rule1['destination_ip_address'] = cidr
else:
rule1['source_ip_address'] = cidr
return [rule1, rule2, rule3, rule4]
def _translated_cidr(self, cidr):
if cidr is None:
return []
else:
return [{'target_id': cidr,
'target_type': 'IPv4Address'}]
def _validate_rules_translation(self, actual_rules, rule_list, is_ingress):
for index in range(len(rule_list)):
self._validate_rule_translation(
actual_rules[index].get_obj_dict(),
rule_list[index],
is_ingress)
def _validate_rule_translation(self, nsx_rule, fw_rule, is_ingress):
self.assertEqual(fw_rule['id'], nsx_rule['id'])
self.assertEqual(fwaas_callbacks_v2.RULE_NAME_PREFIX +
(fw_rule.get('name') or fw_rule['id']),
nsx_rule['display_name'])
self.assertEqual(fw_rule.get('description'), nsx_rule['description'])
self.assertEqual(consts.IN if is_ingress else consts.OUT,
nsx_rule['direction'])
self.assertEqual(self.plugin.nsxpolicy.tier1.get_path(FAKE_ROUTER_ID),
nsx_rule['scope'][0])
# Action
if (fw_rule['action'] == fwaas_consts.FWAAS_REJECT or
fw_rule['action'] == fwaas_consts.FWAAS_DENY):
self.assertEqual(consts.FW_ACTION_DROP, nsx_rule['action'])
else:
self.assertEqual(consts.FW_ACTION_ALLOW, nsx_rule['action'])
# Service
if fw_rule.get('protocol') in ['tcp', 'udp', 'icmp']:
self.assertEqual(['/infra/services/%s-%s-%s' % (
fw_rule['protocol'], FAKE_ROUTER_ID,
fw_rule['id'])],
nsx_rule['services'])
# Source & destination
if (fw_rule.get('source_ip_address') and
not fw_rule['source_ip_address'].startswith('0.0.0.0')):
self.assertEqual(['/infra/domains/%s/groups/source-%s' % (
self.project_id, fw_rule['id'])],
nsx_rule['source_groups'])
elif not is_ingress:
self.assertEqual(['/infra/domains/%s/groups/%s-%s' % (
self.project_id, FAKE_ROUTER_ID, FAKE_NET_ID)],
nsx_rule['source_groups'])
if (fw_rule.get('destination_ip_address') and
not fw_rule['destination_ip_address'].startswith('0.0.0.0')):
self.assertEqual(['/infra/domains/%s/groups/destination-%s' % (
self.project_id, fw_rule['id'])],
nsx_rule['destination_groups'])
elif is_ingress:
self.assertEqual(['/infra/domains/%s/groups/%s-%s' % (
self.project_id, FAKE_ROUTER_ID, FAKE_NET_ID)],
nsx_rule['destination_groups'])
def _fake_empty_firewall_group(self):
fw_inst = {'id': FAKE_FW_ID,
'admin_state_up': True,
'tenant_id': self.project_id,
'ingress_rule_list': [],
'egress_rule_list': []}
return fw_inst
def _fake_firewall_group(self, rule_list, is_ingress=True,
admin_state_up=True):
_rule_list = copy.deepcopy(rule_list)
for rule in _rule_list:
rule['position'] = str(_rule_list.index(rule))
fw_inst = {'id': FAKE_FW_ID,
'admin_state_up': admin_state_up,
'tenant_id': self.project_id,
'ingress_rule_list': [],
'egress_rule_list': []}
if is_ingress:
fw_inst['ingress_rule_list'] = _rule_list
else:
fw_inst['egress_rule_list'] = _rule_list
return fw_inst
def _fake_firewall_group_with_admin_down(self, rule_list,
is_ingress=True):
return self._fake_firewall_group(
rule_list, is_ingress=is_ingress, admin_state_up=False)
def _fake_apply_list(self):
router_inst = {'id': FAKE_ROUTER_ID, 'external_gateway_info': 'dummy'}
router_info_inst = mock.Mock()
router_info_inst.router = router_inst
router_info_inst.router_id = FAKE_ROUTER_ID
apply_list = [(router_info_inst, FAKE_PORT_ID)]
return apply_list
def test_create_firewall_no_rules(self):
apply_list = self._fake_apply_list()
firewall = self._fake_empty_firewall_group()
port = {'id': FAKE_PORT_ID, 'network_id': FAKE_NET_ID}
with mock.patch.object(self.plugin, '_get_router_interfaces',
return_value=[port]),\
mock.patch.object(self.plugin.fwaas_callbacks, 'get_port_fwg',
return_value=firewall),\
mock.patch.object(self.plugin, '_get_router',
return_value={'project_id': self.project_id}),\
mock.patch.object(self.plugin, 'service_router_has_services',
return_value=True),\
mock.patch(GW_POLICY_PATH + ".update_entries") as update_fw:
self.firewall.create_firewall_group('nsx', apply_list, firewall)
# expecting 2 block rules for the logical switch (egress & ingress)
# and last default allow all rule
expected_rules = (self._block_interface_rules(0) +
[self._default_rule(2)])
update_fw.assert_called_once_with(
self.project_id, FAKE_ROUTER_ID, mock.ANY)
# compare rules one by one
actual_rules = update_fw.call_args[0][2]
self.assertEqual(len(expected_rules), len(actual_rules))
for index in range(len(actual_rules)):
self.assertEqual(expected_rules[index],
actual_rules[index].get_obj_dict())
def _setup_firewall_with_rules(self, func, is_ingress=True,
is_conflict=False, cidr='10.24.4.0/24'):
apply_list = self._fake_apply_list()
rule_list = self._fake_rules_v4(is_ingress=is_ingress,
is_conflict=is_conflict,
cidr=cidr)
firewall = self._fake_firewall_group(rule_list, is_ingress=is_ingress)
port = {'id': FAKE_PORT_ID, 'network_id': FAKE_NET_ID}
with mock.patch.object(self.plugin, '_get_router_interfaces',
return_value=[port]),\
mock.patch.object(self.plugin.fwaas_callbacks, 'get_port_fwg',
return_value=firewall), \
mock.patch.object(self.plugin, '_get_router',
return_value={'project_id': self.project_id}),\
mock.patch.object(self.plugin, 'service_router_has_services',
return_value=True), \
mock.patch(GW_POLICY_PATH + ".update_entries") as update_fw:
func('nsx', apply_list, firewall)
expected_default_rules = self._block_interface_rules(
len(rule_list)) + [self._default_rule(len(rule_list) + 2)]
update_fw.assert_called_once_with(
self.project_id, FAKE_ROUTER_ID, mock.ANY)
# compare rules one by one
actual_rules = update_fw.call_args[0][2]
self.assertEqual(len(rule_list) + 3, len(actual_rules))
self._validate_rules_translation(
actual_rules,
rule_list,
is_ingress)
# compare the last 3 rules (default interface rules +
# default allow rule)
self.assertEqual(actual_rules[-3].get_obj_dict(),
expected_default_rules[0])
self.assertEqual(actual_rules[-2].get_obj_dict(),
expected_default_rules[1])
self.assertEqual(actual_rules[-1].get_obj_dict(),
expected_default_rules[2])
def test_create_firewall_with_ingress_rules(self):
self._setup_firewall_with_rules(self.firewall.create_firewall_group)
def test_update_firewall_with_ingress_rules(self):
self._setup_firewall_with_rules(self.firewall.update_firewall_group)
def test_create_firewall_with_egress_rules(self):
self._setup_firewall_with_rules(self.firewall.create_firewall_group,
is_ingress=False)
def test_update_firewall_with_egress_rules(self):
self._setup_firewall_with_rules(self.firewall.update_firewall_group,
is_ingress=False)
def test_create_firewall_with_egress_conflicting_rules(self):
self._setup_firewall_with_rules(self.firewall.update_firewall_group,
is_ingress=False, is_conflict=True)
def test_create_firewall_with_ingress_conflicting_rules(self):
self._setup_firewall_with_rules(self.firewall.update_firewall_group,
is_ingress=True, is_conflict=True)
def test_create_firewall_with_illegal_cidr(self):
self._setup_firewall_with_rules(self.firewall.create_firewall_group,
cidr='0.0.0.0/24')
def test_delete_firewall(self):
apply_list = self._fake_apply_list()
firewall = self._fake_empty_firewall_group()
port = {'id': FAKE_PORT_ID}
with mock.patch.object(self.plugin, '_get_router_interfaces',
return_value=[port]),\
mock.patch.object(self.plugin.fwaas_callbacks, 'get_port_fwg',
return_value=None), \
mock.patch.object(self.plugin, '_get_router',
return_value={'project_id': self.project_id}),\
mock.patch.object(self.plugin, 'service_router_has_services',
return_value=True), \
mock.patch(GW_POLICY_PATH + ".update_entries") as update_fw:
self.firewall.delete_firewall_group('nsx', apply_list, firewall)
# expecting only the default allow-all rule
expected_rules = [self._default_rule(0)]
update_fw.assert_called_once_with(
self.project_id, FAKE_ROUTER_ID, mock.ANY)
# compare rules one by one
actual_rules = update_fw.call_args[0][2]
self.assertEqual(len(expected_rules), len(actual_rules))
for index in range(len(actual_rules)):
self.assertEqual(expected_rules[index],
actual_rules[index].get_obj_dict())
def test_create_firewall_with_admin_down(self):
apply_list = self._fake_apply_list()
rule_list = self._fake_rules_v4()
firewall = self._fake_firewall_group_with_admin_down(rule_list)
with mock.patch.object(self.plugin, 'service_router_has_services',
return_value=True), \
mock.patch.object(self.plugin, '_get_router',
return_value={'project_id': self.project_id}),\
mock.patch(GW_POLICY_PATH + ".update_entries") as update_fw:
self.firewall.create_firewall_group('nsx', apply_list, firewall)
# expecting only the default allow-all rule
expected_rules = [self._default_rule(0)]
update_fw.assert_called_once_with(
self.project_id, FAKE_ROUTER_ID, mock.ANY)
# compare rules one by one
actual_rules = update_fw.call_args[0][2]
self.assertEqual(len(expected_rules), len(actual_rules))
for index in range(len(actual_rules)):
self.assertEqual(expected_rules[index],
actual_rules[index].get_obj_dict())

3
vmware_nsx/tests/unit/nsx_v3/test_fwaas_v2_driver.py
View File

@ -19,7 +19,6 @@ import mock
from neutron_lib.plugins import directory
from vmware_nsx.services.fwaas.nsx_v3 import edge_fwaas_driver_base
from vmware_nsx.services.fwaas.nsx_v3 import edge_fwaas_driver_v2
from vmware_nsx.services.fwaas.nsx_v3 import fwaas_callbacks_v2
from vmware_nsx.tests.unit.nsx_v3 import test_plugin as test_v3_plugin
@ -34,7 +33,7 @@ MOCK_NSX_ID = 'nsx_nsx_router_id'
MOCK_DEFAULT_RULE_ID = 'nsx_default_rule_id'
MOCK_SECTION_ID = 'sec_id'
DEFAULT_RULE = {'is_default': True,
'display_name': edge_fwaas_driver_base.DEFAULT_RULE_NAME,
'display_name': edge_fwaas_driver_v2.DEFAULT_RULE_NAME,
'id': MOCK_DEFAULT_RULE_ID,
'action': consts.FW_ACTION_DROP}