x/vmware-nsx
Code Issues Proposed changes

Merge "NSX|V3 refactor fwaas to support plugin rules"

Browse Source
This commit is contained in:
Jenkins 2017-10-04 22:15:39 +00:00 committed by Gerrit Code Review
commit ab048f384c
14 changed files with 349 additions and 283 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
doc/source/devstack.rst
View File

@ -195,7 +195,7 @@ Add neutron-fwaas repo as an external repository and configure following flags i
[fwaas]
enabled = True
driver = vmware_nsxv3_edge
driver = vmware_nsxv3_edge_v1
FWaaS (V2) Driver

3
setup.cfg
View File

@ -35,7 +35,8 @@ neutron.core_plugins =
vmware_dvs = vmware_nsx.plugin:NsxDvsPlugin
firewall_drivers =
vmware_nsxv_edge = vmware_nsx.services.fwaas.nsx_v.edge_fwaas_driver:EdgeFwaasDriver
vmware_nsxv3_edge = vmware_nsx.services.fwaas.nsx_v3.edge_fwaas_driver:EdgeFwaasV3Driver
vmware_nsxv3_edge = vmware_nsx.services.fwaas.nsx_v3.edge_fwaas_driver_v1:EdgeFwaasV3DriverV1
vmware_nsxv3_edge_v1 = vmware_nsx.services.fwaas.nsx_v3.edge_fwaas_driver_v1:EdgeFwaasV3DriverV1
vmware_nsxv3_edge_v2 = vmware_nsx.services.fwaas.nsx_v3.edge_fwaas_driver_v2:EdgeFwaasV3DriverV2
neutron.service_plugins =
vmware_nsxv_qos = vmware_nsx.services.qos.nsx_v.plugin:NsxVQosPlugin

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

@ -97,7 +97,7 @@ from vmware_nsx.plugins.common import plugin as nsx_plugin_common
from vmware_nsx.plugins.nsx_v3 import availability_zones as nsx_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_v3 import fwaas_callbacks
from vmware_nsx.services.fwaas.nsx_v3 import fwaas_callbacks_v1
from vmware_nsx.services.fwaas.nsx_v3 import fwaas_callbacks_v2
from vmware_nsx.services.lbaas.nsx_v3 import lb_driver_v2
from vmware_nsx.services.qos.common import utils as qos_com_utils
@ -278,15 +278,14 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
})
def _init_fwaas(self, resource, event, trigger, **kwargs):
self.fwaas_callbacks_v1 = None
self.fwaas_callbacks_v2 = None
self.fwaas_callbacks = None
if fwaas_utils.is_fwaas_v1_plugin_enabled():
LOG.info("NSXv3 FWaaS v1 plugin enabled")
self.fwaas_callbacks_v1 = fwaas_callbacks.Nsxv3FwaasCallbacks(
self.fwaas_callbacks = fwaas_callbacks_v1.Nsxv3FwaasCallbacksV1(
self.nsxlib)
if fwaas_utils.is_fwaas_v2_plugin_enabled():
LOG.info("NSXv3 FWaaS v2 plugin enabled")
self.fwaas_callbacks_v2 = fwaas_callbacks_v2.Nsxv3FwaasCallbacksV2(
self.fwaas_callbacks = fwaas_callbacks_v2.Nsxv3FwaasCallbacksV2(
self.nsxlib)
def _init_lbv2_driver(self):
@ -3291,41 +3290,35 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
def update_router_firewall(self, context, router_id):
"""Rewrite all the rules in the router edge firewall
Currently only for FWaaS v2
This method should be called on FWaaS v2 updates, and on router
This method should be called on FWaaS v1/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.
"""
# make sure fwaas v2 is enabled
if (not self.fwaas_callbacks_v2 or
not self.fwaas_callbacks_v2.fwaas_enabled):
return
fwaas_callbacks = self.fwaas_callbacks_v2
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)
# find the backend router and its firewall section
nsx_id, sect_id = fwaas_callbacks.get_backend_router_and_fw_section(
context, router_id)
# let the fwaas callbacks update the router FW
return self.fwaas_callbacks.update_router_firewall(
context, self.nsxlib, router_id, ports)
# find all the relevant ports of the router
# TODO(asarfaty): Add vm ports as well
ports = self._get_router_interfaces(context, router_id)
def get_extra_fw_rules(self, context, router_id, port_id=None):
"""Return firewall rules that should be added to the router firewall
fw_rules = []
for port in ports:
_net_id, nsx_port_id = nsx_db.get_nsx_switch_and_port_id(
context.session, port['id'])
# add the rules for this port, only if it has an active fw
fwg = fwaas_callbacks.get_port_fwg(context, port['id'])
if fwg:
port_rules = fwaas_callbacks.get_port_rules(nsx_port_id, fwg)
fw_rules.extend(port_rules)
# add a default allow-all rule to all other traffic
fw_rules.append(fwaas_callbacks.get_default_allow_all_rule(
sect_id))
# update the backend
self.nsxlib.firewall_section.update(sect_id, rules=fw_rules)
This method should return a list of allow firewall rules that are
required in order to enable different plugin features with north/south
traffic.
The returned rules will be added after the FWaaS rules, and before the
default drop rule.
if port_id is specified, only rules relevant for this router interface
port should be returned, and the rules should be ingress/egress
(but not both) and include the source/dest nsx logical port.
"""
#TODO(asarfaty): DHCP relay rules
return []
def _get_ports_and_address_groups(self, context, router_id, network_id,
exclude_sub_ids=None):

6
vmware_nsx/services/fwaas/common/fwaas_callbacks.py → vmware_nsx/services/fwaas/common/fwaas_callbacks_v1.py
View File

@ -130,3 +130,9 @@ class NsxFwaasCallbacks(firewall_l3_agent.L3WithFWaaS):
for fw in fw_list:
if fw['id'] == fw_id:
return fw
def get_router_firewall(self, context, router_id):
ctx_elevated = context.elevated()
fw_id = self._get_router_firewall_id(ctx_elevated, router_id)
if fw_id:
return self._get_fw_from_plugin(ctx_elevated, fw_id)

4
vmware_nsx/services/fwaas/nsx_v/fwaas_callbacks.py
View File

@ -15,12 +15,12 @@
from oslo_log import log as logging
from vmware_nsx.services.fwaas.common import fwaas_callbacks as com_callbacks
from vmware_nsx.services.fwaas.common import fwaas_callbacks_v1 as com_clbcks
LOG = logging.getLogger(__name__)
class NsxvFwaasCallbacks(com_callbacks.NsxFwaasCallbacks):
class NsxvFwaasCallbacks(com_clbcks.NsxFwaasCallbacks):
"""NSX-V RPC callbacks for Firewall As A Service - V1."""
def should_apply_firewall_to_router(self, context, router, router_id):

176
vmware_nsx/services/fwaas/nsx_v3/edge_fwaas_driver.py
View File

@ -1,176 +0,0 @@
# Copyright 2017 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 import context as n_context
from oslo_log import helpers as log_helpers
from oslo_log import log as logging
from neutron_lib.exceptions import firewall_v1 as exceptions
from vmware_nsx.services.fwaas.nsx_v3 import edge_fwaas_driver_base as \
base_driver
LOG = logging.getLogger(__name__)
FWAAS_DRIVER_NAME = 'Fwaas V1 NSX-V3 driver'
NSX_FW_TAG = 'os-neutron-fw-id'
class EdgeFwaasV3Driver(base_driver.CommonEdgeFwaasV3Driver):
"""NSX-V3 driver for Firewall As A Service - V1."""
def __init__(self):
exception_cls = exceptions.FirewallInternalDriverError
super(EdgeFwaasV3Driver, self).__init__(exception_cls,
FWAAS_DRIVER_NAME)
def _create_or_update_firewall(self, agent_mode, apply_list, firewall):
# admin state down means default block rule firewall
if not firewall['admin_state_up']:
self.apply_default_policy(agent_mode, apply_list, firewall)
return
context = n_context.get_admin_context()
rules = self._translate_rules(firewall['firewall_rule_list'])
# update each router on the backend
self._update_backend_routers(context, apply_list, firewall['id'],
rules=rules)
@log_helpers.log_method_call
def create_firewall(self, agent_mode, apply_list, firewall):
"""Create the Firewall with a given policy. """
self.validate_backend_version()
self._create_or_update_firewall(agent_mode, apply_list, firewall)
@log_helpers.log_method_call
def update_firewall(self, agent_mode, apply_list, firewall):
"""Remove previous policy and apply the new policy."""
self.validate_backend_version()
self._create_or_update_firewall(agent_mode, apply_list, firewall)
@log_helpers.log_method_call
def delete_firewall(self, agent_mode, apply_list, firewall):
"""Delete firewall.
Removes rules created by this instance from the backend firewall
And add the default allow rule.
"""
self.validate_backend_version()
context = n_context.get_admin_context()
self._update_backend_routers(context, apply_list, firewall['id'],
delete_fw=True)
@log_helpers.log_method_call
def apply_default_policy(self, agent_mode, apply_list, firewall):
"""Apply the default policy (deny all).
The backend firewall always has this policy (=deny all) as default,
so we only need to delete the current rules.
"""
self.validate_backend_version()
context = n_context.get_admin_context()
self._update_backend_routers(context, apply_list, firewall['id'],
rules=[])
def _update_backend_routers(self, context, apply_list, fw_id, rules=None,
delete_fw=False):
# update each router on the backend
for router_info in apply_list:
# Skip unsupported routers
if not self.should_apply_firewall_to_router(router_info.router):
continue
router_id = router_info.router_id
# update the routers firewall
if delete_fw:
self._delete_nsx_router_firewall(context, router_id)
else:
self._update_nsx_router_firewall(context, router_id, fw_id,
rules)
def _update_nsx_router_tags(self, nsx_router_id, fw_id=None):
"""Get the updated tags to put on the nsx-router
With/without the firewall id
"""
# Get the current tags
nsx_router = self.nsx_router.get(nsx_router_id)
if 'tags' not in nsx_router:
nsx_router['tags'] = []
tags = nsx_router['tags']
# Look for the firewall tag and update/remove it
update_tags = False
found_tag = False
for tag in tags:
if tag.get('scope') == NSX_FW_TAG:
found_tag = True
if not fw_id:
tags.remove(tag)
update_tags = True
break
if fw_id != tag.get('tag'):
tag['tag'] = fw_id
update_tags = True
break
# Add the tag if not found
if fw_id and not found_tag:
tags.append({'scope': NSX_FW_TAG,
'tag': fw_id})
update_tags = True
# update tags on the backend router
if update_tags:
self.nsx_router.update(nsx_router_id, tags=tags)
def _delete_nsx_router_firewall(self, context, router_id):
"""Reset the router firewall back to it's default"""
# find the backend router and its firewall section
nsx_router_id, section_id = self.get_backend_router_and_fw_section(
context, router_id)
# Add default allow all rule
allow_all = self.get_default_backend_rule(section_id, allow_all=True)
# Update the backend firewall section with the rules
self.nsx_firewall.update(section_id, rules=[allow_all])
# Also update the router tags
self._update_nsx_router_tags(nsx_router_id)
def _update_nsx_router_firewall(self, context, router_id, fw_id, rules):
"""Update the backend router firewall section
Adding all relevant north-south rules from the FWaaS firewall
and the default drop all rule
Since those rules do no depend on the router gateway/interfaces/ips
there is no need to call this method on each router update.
Just when the firewall changes.
"""
# find the backend router and its firewall section
nsx_router_id, section_id = self.get_backend_router_and_fw_section(
context, router_id)
#TODO(asarfaty) add dhcp relay allow rules here
# Add default drop all rule at the end
drop_all = self.get_default_backend_rule(section_id, allow_all=False)
# Update the backend firewall section with the rules
self.nsx_firewall.update(section_id, rules=rules + [drop_all])
# Also update the router tags
self._update_nsx_router_tags(nsx_router_id, fw_id=fw_id)

119
vmware_nsx/services/fwaas/nsx_v3/edge_fwaas_driver_v1.py Normal file
View File

@ -0,0 +1,119 @@
# Copyright 2017 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 import context as n_context
from oslo_log import helpers as log_helpers
from oslo_log import log as logging
from neutron_lib.exceptions import firewall_v1 as exceptions
from vmware_nsx.services.fwaas.nsx_v3 import edge_fwaas_driver_base as \
base_driver
LOG = logging.getLogger(__name__)
FWAAS_DRIVER_NAME = 'Fwaas V1 NSX-V3 driver'
NSX_FW_TAG = 'os-neutron-fw-id'
class EdgeFwaasV3DriverV1(base_driver.CommonEdgeFwaasV3Driver):
"""NSX-V3 driver for Firewall As A Service - V1."""
def __init__(self):
exception_cls = exceptions.FirewallInternalDriverError
super(EdgeFwaasV3DriverV1, self).__init__(exception_cls,
FWAAS_DRIVER_NAME)
@log_helpers.log_method_call
def create_firewall(self, agent_mode, apply_list, firewall):
"""Create the Firewall with a given policy. """
self._update_backend_routers(apply_list, firewall['id'])
@log_helpers.log_method_call
def update_firewall(self, agent_mode, apply_list, firewall):
"""Remove previous policy and apply the new policy."""
self._update_backend_routers(apply_list, firewall['id'])
@log_helpers.log_method_call
def delete_firewall(self, agent_mode, apply_list, firewall):
"""Delete firewall.
Removes rules created by this instance from the backend firewall
And add the default allow rule.
"""
self._update_backend_routers(apply_list, firewall['id'])
@log_helpers.log_method_call
def apply_default_policy(self, agent_mode, apply_list, firewall):
"""Apply the default policy (deny all).
The backend firewall always has this policy (=deny all) as default,
so we only need to delete the current rules.
"""
self._update_backend_routers(apply_list, firewall['id'])
def _update_backend_routers(self, apply_list, fw_id):
""""Update each router on the backend using the core plugin code"""
self.validate_backend_version()
context = n_context.get_admin_context()
for router_info in apply_list:
# Skip unsupported routers
if not self.should_apply_firewall_to_router(router_info.router):
continue
self.core_plugin.update_router_firewall(
context, router_info.router_id)
def update_nsx_router_tags(self, nsx_router_id, fw_id=None):
"""Update the backend router with tags marking the attached fw id"""
# Get the current tags
nsx_router = self.nsx_router.get(nsx_router_id)
if 'tags' not in nsx_router:
nsx_router['tags'] = []
tags = nsx_router['tags']
# Look for the firewall tag and update/remove it
update_tags = False
found_tag = False
for tag in tags:
if tag.get('scope') == NSX_FW_TAG:
found_tag = True
if not fw_id:
tags.remove(tag)
update_tags = True
break
if fw_id != tag.get('tag'):
tag['tag'] = fw_id
update_tags = True
break
# Add the tag if not found
if fw_id and not found_tag:
tags.append({'scope': NSX_FW_TAG,
'tag': fw_id})
update_tags = True
# update tags on the backend router
if update_tags:
self.nsx_router.update(nsx_router_id, tags=tags)
def get_router_translated_rules(self, router_id, firewall):
"""Return the list of translated rules
The default drop all will be added later
"""
# Return the firewall rules only if the fw is up
if firewall['admin_state_up']:
return self._translate_rules(firewall['firewall_rule_list'])
return []

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

@ -82,7 +82,8 @@ class EdgeFwaasV3DriverV2(base_driver.CommonEdgeFwaasV3Driver):
for router_id in routers:
self.core_plugin.update_router_firewall(context, router_id)
def get_port_translated_rules(self, nsx_port_id, firewall_group):
def get_port_translated_rules(self, nsx_port_id, firewall_group,
plugin_rules):
"""Return the list of translated rules per port"""
port_rules = []
# Add the firewall group ingress/egress rules only if the fw is up
@ -94,6 +95,10 @@ class EdgeFwaasV3DriverV2(base_driver.CommonEdgeFwaasV3Driver):
firewall_group['egress_rule_list'],
replace_src=nsx_port_id))
# Add the per-port plugin rules
if plugin_rules and isinstance(plugin_rules, list):
port_rules.extend(plugin_rules)
# Add ingress/egress block rules for this port
port_rules.extend([
{'display_name': "Block port ingress",

47
vmware_nsx/services/fwaas/nsx_v3/fwaas_callbacks.py
View File

@ -1,47 +0,0 @@
# Copyright 2017 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 vmware_nsx.services.fwaas.common import fwaas_callbacks as com_callbacks
LOG = logging.getLogger(__name__)
class Nsxv3FwaasCallbacks(com_callbacks.NsxFwaasCallbacks):
"""NSX-V3 RPC callbacks for Firewall As A Service - V1."""
def __init__(self, nsxlib):
super(Nsxv3FwaasCallbacks, self).__init__()
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(Nsxv3FwaasCallbacks,
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.fwaas_driver.should_apply_firewall_to_router(router_data):
return False
return True

88
vmware_nsx/services/fwaas/nsx_v3/fwaas_callbacks_v1.py Normal file
View File

@ -0,0 +1,88 @@
# Copyright 2017 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 vmware_nsx.services.fwaas.common import fwaas_callbacks_v1 as com_clbcks
LOG = logging.getLogger(__name__)
class Nsxv3FwaasCallbacksV1(com_clbcks.NsxFwaasCallbacks):
"""NSX-V3 RPC callbacks for Firewall As A Service - V1."""
def __init__(self, nsxlib):
super(Nsxv3FwaasCallbacksV1, self).__init__()
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(Nsxv3FwaasCallbacksV1,
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.fwaas_driver.should_apply_firewall_to_router(router_data):
return False
return True
def update_router_firewall(self, context, nsxlib, router_id,
router_interfaces):
"""Rewrite all the FWaaS v1 rules in the router edge firewall
This method should be called on FWaaS updates, and on router
interfaces changes.
"""
# find the backend router and its firewall section
nsx_id, sect_id = self.fwaas_driver.get_backend_router_and_fw_section(
context, router_id)
fw_rules = []
fw_id = None
if self.should_apply_firewall_to_router(context, router_id):
# Find the firewall attached to this router
# (must have one since should_apply returned true)
firewall = self.get_router_firewall(context, router_id)
fw_id = firewall['id']
# Add the FW rules
fw_rules.extend(self.fwaas_driver.get_router_translated_rules(
router_id, firewall))
# Add plugin additional allow rules
fw_rules.extend(self.core_plugin.get_extra_fw_rules(
context, router_id))
# Add the default drop all rule
fw_rules.append(self.fwaas_driver.get_default_backend_rule(
sect_id, allow_all=False))
else:
# default allow all rule
fw_rules.append(self.fwaas_driver.get_default_backend_rule(
sect_id, allow_all=True))
# update the backend
nsxlib.firewall_section.update(sect_id, rules=fw_rules)
# Also update the router tags
self.fwaas_driver.update_nsx_router_tags(nsx_id, fw_id=fw_id)

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

@ -15,6 +15,7 @@
from oslo_log import log as logging
from vmware_nsx.db import db as nsx_db
from vmware_nsx.services.fwaas.common import fwaas_callbacks_v2 as \
com_callbacks
@ -47,14 +48,43 @@ class Nsxv3FwaasCallbacksV2(com_callbacks.NsxFwaasCallbacksV2):
return True
def get_port_rules(self, nsx_port_id, fwg):
return self.fwaas_driver.get_port_translated_rules(nsx_port_id, fwg)
def get_port_rules(self, nsx_port_id, fwg, plugin_rules):
return self.fwaas_driver.get_port_translated_rules(nsx_port_id, fwg,
plugin_rules)
def get_backend_router_and_fw_section(self, context, router_id):
"""Find the backend router and its firewall section"""
return self.fwaas_driver.get_backend_router_and_fw_section(
def update_router_firewall(self, context, nsxlib, router_id,
router_interfaces):
"""Rewrite all the FWaaS v2 rules in the router edge firewall
This method should be called on FWaaS updates, and on router
interfaces changes.
"""
# find the backend router and its firewall section
nsx_id, sect_id = self.fwaas_driver.get_backend_router_and_fw_section(
context, router_id)
def get_default_allow_all_rule(self, section_id):
return self.fwaas_driver.get_default_backend_rule(
section_id, allow_all=True)
fw_rules = []
# Add firewall rules per port attached to a firewall group
for port in router_interfaces:
_net_id, nsx_port_id = nsx_db.get_nsx_switch_and_port_id(
context.session, port['id'])
# Check if this port has a firewall
fwg = self.get_port_fwg(context, port['id'])
if fwg:
# Add plugin additional allow rules
plugin_rules = self.core_plugin.get_extra_fw_rules(
context, router_id, port['id'])
# 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_rules(nsx_port_id, fwg,
plugin_rules))
# add a default allow-all rule to all other traffic & ports
fw_rules.append(self.fwaas_driver.get_default_backend_rule(
sect_id, allow_all=True))
# update the backend router firewall
nsxlib.firewall_section.update(sect_id, rules=fw_rules)

57
vmware_nsx/tests/unit/nsx_v3/test_fwaas_driver.py → vmware_nsx/tests/unit/nsx_v3/test_fwaas_v1_driver.py
View File

@ -16,11 +16,15 @@
import copy
import mock
from vmware_nsxlib.v3 import nsx_constants as consts
from vmware_nsx.services.fwaas.nsx_v3 import edge_fwaas_driver
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_v1 as \
edge_fwaas_driver
from vmware_nsx.services.fwaas.nsx_v3 import fwaas_callbacks_v1
from vmware_nsx.tests.unit.nsx_v3 import test_plugin as test_v3_plugin
from vmware_nsxlib.v3 import nsx_constants as consts
FAKE_FW_ID = 'fake_fw_uuid'
FAKE_ROUTER_ID = 'fake_rtr_uuid'
@ -36,7 +40,7 @@ DEFAULT_RULE = {'is_default': True,
class Nsxv3FwaasTestCase(test_v3_plugin.NsxV3PluginTestCaseMixin):
def setUp(self):
super(Nsxv3FwaasTestCase, self).setUp()
self.firewall = edge_fwaas_driver.EdgeFwaasV3Driver()
self.firewall = edge_fwaas_driver.EdgeFwaasV3DriverV1()
# Start some nsxlib/DB mocks
mock.patch(
@ -53,6 +57,12 @@ class Nsxv3FwaasTestCase(test_v3_plugin.NsxV3PluginTestCaseMixin):
"vmware_nsx.db.db.get_nsx_router_id",
return_value=MOCK_NSX_ID).start()
self.plugin = directory.get_plugin()
self.plugin.fwaas_callbacks = fwaas_callbacks_v1.\
Nsxv3FwaasCallbacksV1(self.plugin.nsxlib)
self.plugin.fwaas_callbacks.fwaas_enabled = True
self.plugin.fwaas_callbacks.fwaas_driver = self.firewall
def _default_rule(self, drop=True):
rule = DEFAULT_RULE
if drop:
@ -164,7 +174,17 @@ class Nsxv3FwaasTestCase(test_v3_plugin.NsxV3PluginTestCaseMixin):
rule_list = self._fake_rules_v4()
firewall = self._fake_firewall(rule_list)
with mock.patch("vmware_nsxlib.v3.security.NsxLibFirewallSection."
"update") as update_fw:
"update") as update_fw, \
mock.patch.object(self.plugin, '_get_router_interfaces',
return_value=[]), \
mock.patch.object(self.plugin, 'get_router',
return_value=apply_list[0]), \
mock.patch.object(self.plugin.fwaas_callbacks,
'_get_router_firewall_id',
return_value=firewall['id']), \
mock.patch.object(self.plugin.fwaas_callbacks,
'_get_fw_from_plugin',
return_value=firewall):
func('nsx', apply_list, firewall)
self.assertEqual(router_count, update_fw.call_count)
update_fw.assert_called_with(
@ -177,6 +197,16 @@ class Nsxv3FwaasTestCase(test_v3_plugin.NsxV3PluginTestCaseMixin):
initial_tags = [{'scope': 'xxx', 'tag': 'yyy'}]
with mock.patch("vmware_nsxlib.v3.security.NsxLibFirewallSection."
"update") as update_fw,\
mock.patch.object(self.plugin, '_get_router_interfaces',
return_value=[]), \
mock.patch.object(self.plugin, 'get_router',
return_value=apply_list[0]), \
mock.patch.object(self.plugin.fwaas_callbacks,
'_get_router_firewall_id',
return_value=firewall['id']), \
mock.patch.object(self.plugin.fwaas_callbacks,
'_get_fw_from_plugin',
return_value=firewall), \
mock.patch("vmware_nsxlib.v3.core_resources.NsxLibLogicalRouter."
"update") as update_rtr,\
mock.patch("vmware_nsxlib.v3.core_resources.NsxLibLogicalRouter."
@ -209,6 +239,13 @@ class Nsxv3FwaasTestCase(test_v3_plugin.NsxV3PluginTestCaseMixin):
'tag': firewall['id']}]
with mock.patch("vmware_nsxlib.v3.security.NsxLibFirewallSection."
"update") as update_fw,\
mock.patch.object(self.plugin, '_get_router_interfaces',
return_value=[]), \
mock.patch.object(self.plugin, 'get_router',
return_value=apply_list[0]), \
mock.patch.object(self.plugin.fwaas_callbacks,
'_get_router_firewall_id',
return_value=None), \
mock.patch("vmware_nsxlib.v3.core_resources.NsxLibLogicalRouter."
"update") as update_rtr,\
mock.patch("vmware_nsxlib.v3.core_resources.NsxLibLogicalRouter."
@ -229,7 +266,17 @@ class Nsxv3FwaasTestCase(test_v3_plugin.NsxV3PluginTestCaseMixin):
rule_list = self._fake_rules_v4()
firewall = self._fake_firewall_with_admin_down(rule_list)
with mock.patch("vmware_nsxlib.v3.security.NsxLibFirewallSection."
"update") as update_fw:
"update") as update_fw, \
mock.patch.object(self.plugin, '_get_router_interfaces',
return_value=[]), \
mock.patch.object(self.plugin, 'get_router',
return_value=apply_list[0]), \
mock.patch.object(self.plugin.fwaas_callbacks,
'_get_router_firewall_id',
return_value=firewall['id']), \
mock.patch.object(self.plugin.fwaas_callbacks,
'_get_fw_from_plugin',
return_value=firewall):
self.firewall.create_firewall('nsx', apply_list, firewall)
update_fw.assert_called_once_with(
MOCK_SECTION_ID,

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

@ -60,10 +60,10 @@ class Nsxv3FwaasTestCase(test_v3_plugin.NsxV3PluginTestCaseMixin):
return_value=MOCK_NSX_ID).start()
self.plugin = directory.get_plugin()
self.plugin.fwaas_callbacks_v2 = fwaas_callbacks_v2.\
self.plugin.fwaas_callbacks = fwaas_callbacks_v2.\
Nsxv3FwaasCallbacksV2(self.plugin.nsxlib)
self.plugin.fwaas_callbacks_v2.fwaas_enabled = True
self.plugin.fwaas_callbacks_v2.fwaas_driver = self.firewall
self.plugin.fwaas_callbacks.fwaas_enabled = True
self.plugin.fwaas_callbacks.fwaas_driver = self.firewall
def _default_rule(self):
rule = DEFAULT_RULE
@ -194,7 +194,7 @@ class Nsxv3FwaasTestCase(test_v3_plugin.NsxV3PluginTestCaseMixin):
port = {'id': FAKE_PORT_ID}
with mock.patch.object(self.plugin, '_get_router_interfaces',
return_value=[port]),\
mock.patch.object(self.plugin.fwaas_callbacks_v2, 'get_port_fwg',
mock.patch.object(self.plugin.fwaas_callbacks, 'get_port_fwg',
return_value=firewall),\
mock.patch("vmware_nsx.db.db.get_nsx_switch_and_port_id",
return_value=(0, FAKE_NSX_PORT_ID)),\
@ -227,7 +227,7 @@ class Nsxv3FwaasTestCase(test_v3_plugin.NsxV3PluginTestCaseMixin):
port = {'id': FAKE_PORT_ID}
with mock.patch.object(self.plugin, '_get_router_interfaces',
return_value=[port]),\
mock.patch.object(self.plugin.fwaas_callbacks_v2, 'get_port_fwg',
mock.patch.object(self.plugin.fwaas_callbacks, 'get_port_fwg',
return_value=firewall),\
mock.patch("vmware_nsx.db.db.get_nsx_switch_and_port_id",
return_value=(0, FAKE_NSX_PORT_ID)),\
@ -281,7 +281,7 @@ class Nsxv3FwaasTestCase(test_v3_plugin.NsxV3PluginTestCaseMixin):
port = {'id': FAKE_PORT_ID}
with mock.patch.object(self.plugin, '_get_router_interfaces',
return_value=[port]),\
mock.patch.object(self.plugin.fwaas_callbacks_v2, 'get_port_fwg',
mock.patch.object(self.plugin.fwaas_callbacks, 'get_port_fwg',
return_value=None),\
mock.patch("vmware_nsx.db.db.get_nsx_switch_and_port_id",
return_value=(0, FAKE_NSX_PORT_ID)),\

2
vmware_nsx/tests/unit/nsx_v3/test_plugin.py
View File

@ -905,7 +905,7 @@ class L3NatTest(test_l3_plugin.L3BaseForIntTests, NsxV3PluginTestCaseMixin,
self.plugin_instance.__module__,
self.plugin_instance.__class__.__name__)
self._plugin_class = self.plugin_instance.__class__
self.plugin_instance.fwaas_callbacks_v2 = None
self.plugin_instance.fwaas_callbacks = None
def test_floatingip_create_different_fixed_ip_same_port(self):
self.skipTest('Multiple fixed ips on a port are not supported')