NSX|v3: FWaaS v2 support

FWaaS V2 support in NSX-v3.
Support different firewall group per router interface port for
igress/egress.
limitation: cannot support egress rules with source ip, or ingress
rules with destination ips.

Depends-on: I2a37be5518bfc8124ffca2ab05f684d8c1c3d673
Change-Id: I3ed70fa48d078bed15f30e855b73bdfb11d11c6e
This commit is contained in:
Adit Sarfaty 2017-08-21 13:18:20 +03:00
parent f42880f989
commit 32e95f47b5
13 changed files with 1004 additions and 210 deletions

View File

@ -89,14 +89,14 @@ lines to the policy.json file::
"delete_flow_classifier": "rule:admin_only",
"get_flow_classifier": "rule:admin_only"
FWAAS (V1) Driver
FWaaS (V1) Driver
~~~~~~~~~~~~~~~~~
Add neutron-fwaas repo as an external repository and configure following flags in ``local.conf``::
[[local|localrc]]
enable_plugin neutron-fwaas https://git.openstack.org/openstack/neutron-fwaas
ENABLED_SERVICES+=,q-fwaas
ENABLED_SERVICES+=,q-fwaas-v1
[[post-config|$NEUTRON_CONF]]
[DEFAULT]
@ -106,7 +106,6 @@ Add neutron-fwaas repo as an external repository and configure following flags i
enabled = True
driver = vmware_nsxv_edge
Neutron dynamic routing plugin (bgp)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@ -181,7 +180,7 @@ Enable trunk service and configure following flags in ``local.conf``::
ENABLED_SERVICES+=,q-trunk
Q_SERVICE_PLUGIN_CLASSES=trunk
FWAAS (V1) Driver:
FWaaS (V1) Driver:
~~~~~~~~~~~~~
Add neutron-fwaas repo as an external repository and configure following flags in ``local.conf``::
@ -198,6 +197,24 @@ Add neutron-fwaas repo as an external repository and configure following flags i
enabled = True
driver = vmware_nsxv3_edge
FWaaS (V2) Driver
~~~~~~~~~~~~~~~~~
Add neutron-fwaas repo as an external repository and configure following flags in ``local.conf``::
[[local|localrc]]
enable_plugin neutron-fwaas https://git.openstack.org/openstack/neutron-fwaas
ENABLED_SERVICES+=,q-fwaas-v2
[[post-config|$NEUTRON_CONF]]
[DEFAULT]
service_plugins = neutron_fwaas.services.firewall.fwaas_plugin_v2.FirewallPluginV2
[fwaas]
enabled = True
driver = vmware_nsxv3_edge_v2
LBaaS v2 Driver
~~~~~~~~~~~~~~~

View File

@ -0,0 +1,6 @@
---
prelude: >
The NSX-v3 plugin supports FWaaS V2.
features:
The NSX-v3 plugin now supports FWaaS V2 allowing to set a
different firewall group policy on each router port.

View File

@ -37,6 +37,7 @@ neutron.core_plugins =
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_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
neutron.qos.notification_drivers =

View File

@ -96,7 +96,9 @@ from vmware_nsx.extensions import securitygrouplogging as sg_logging
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_v2
from vmware_nsx.services.lbaas.nsx_v3 import lb_driver_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
@ -235,7 +237,9 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
self._init_nsx_profiles()
# Init the FWaaS support
self._init_fwaas()
registry.subscribe(
self._init_fwaas,
resources.PROCESS, events.AFTER_INIT)
# Include exclude NSGroup
LOG.debug("Initializing NSX v3 Excluded Port NSGroup")
@ -273,9 +277,17 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
webob.exc.HTTPBadRequest,
})
def _init_fwaas(self):
# Bind FWaaS callbacks to the driver
self.fwaas_callbacks = fwaas_callbacks.Nsxv3FwaasCallbacks(self.nsxlib)
def _init_fwaas(self, resource, event, trigger, **kwargs):
self.fwaas_callbacks_v1 = None
self.fwaas_callbacks_v2 = None
if fwaas_utils.is_fwaas_v1_plugin_enabled():
LOG.info("NSXv3 FWaaS v1 plugin enabled")
self.fwaas_callbacks_v1 = fwaas_callbacks.Nsxv3FwaasCallbacks(
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.nsxlib)
def _init_lbv2_driver(self):
# Get LBaaSv2 driver during plugin initialization. If the platform
@ -3215,6 +3227,45 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
self._routerlib.add_static_routes(nsx_router_id, route)
router_db['status'] = curr_status
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
interfaces changes.
"""
# 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
# find the backend router and its firewall section
nsx_id, sect_id = fwaas_callbacks.get_backend_router_and_fw_section(
context, router_id)
# find all the relevant ports of the router
# TODO(asarfaty): Add vm ports as well
ports = self._get_router_interfaces(context, router_id)
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)
def _get_ports_and_address_groups(self, context, router_id, network_id,
exclude_sub_ids=None):
exclude_sub_ids = [] if not exclude_sub_ids else exclude_sub_ids
@ -3351,6 +3402,8 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
context, gw_network_id)
self._add_subnet_snat_rule(context, router_id, nsx_router_id,
subnet, gw_address_scope, gw_ip)
# update firewall rules
self.update_router_firewall(context, router_id)
except Exception:
with excutils.save_and_reraise_exception():
LOG.error("Neutron failed to add_router_interface on "
@ -3441,6 +3494,10 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
# (with the network) if this is the last DHCP-disabled subnet on
# the router.
nsx_rpc.handle_router_metadata_access(self, context, router_id)
# update firewall rules
self.update_router_firewall(context, router_id)
return info
def _create_floating_ip_wrapper(self, context, floatingip):

View File

@ -0,0 +1,149 @@
# 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_config import cfg
from oslo_log import log as logging
from neutron.agent.l3 import router_info
from neutron.common import config as neutron_config # noqa
from neutron_fwaas.db.firewall.v2 import firewall_db_v2
from neutron_fwaas.services.firewall.agents.l3reference \
import firewall_l3_agent_v2
from neutron_lib import constants as nl_constants
from neutron_lib import context as n_context
from neutron_lib.plugins import directory
LOG = logging.getLogger(__name__)
class DummyAgentApi(object):
def is_router_in_namespace(self, router_id):
return True
class NsxFwaasCallbacksV2(firewall_l3_agent_v2.L3WithFWaaS):
"""Common NSX RPC callbacks for Firewall As A Service - V2."""
def __init__(self):
# The super code needs a configuration object with the neutron host
# and an agent_mode, which our driver doesn't use.
neutron_conf = cfg.CONF
neutron_conf.agent_mode = 'nsx'
super(NsxFwaasCallbacksV2, self).__init__(conf=neutron_conf)
self.agent_api = DummyAgentApi()
@property
def core_plugin(self):
return directory.get_plugin()
# Override functions using the agent_api that is not used by our plugin
def _get_firewall_group_ports(self, context, firewall_group,
to_delete=False, require_new_plugin=False):
"""Returns in-namespace ports, either from firewall group dict if newer
version of plugin or from project routers otherwise.
NOTE: Vernacular move from "tenant" to "project" doesn't yet appear
as a key in router or firewall group objects.
"""
fwg_port_ids = []
if self._has_port_insertion_fields(firewall_group):
if to_delete:
fwg_port_ids = firewall_group['del-port-ids']
else:
fwg_port_ids = firewall_group['add-port-ids']
elif not require_new_plugin:
routers = self._get_routers_in_project(
context, firewall_group['tenant_id'])
for router in routers:
if router.router['tenant_id'] == firewall_group['tenant_id']:
fwg_port_ids.extend([p['id'] for p in
router.internal_ports])
# Return in-namespace port objects.
return self._get_in_ns_ports(fwg_port_ids)
def _get_in_ns_ports(self, port_ids):
"""Returns port objects in the local namespace, along with their
router_info.
"""
context = n_context.get_admin_context()
in_ns_ports = {} # This will be converted to a list later.
for port_id in port_ids:
# find the router of this port:
port = self.core_plugin.get_port(context, port_id)
router_id = port['device_id']
router = self.core_plugin.get_router(context, router_id)
router_info = self._router_dict_to_obj(router)
if router_info:
if router_info in in_ns_ports:
in_ns_ports[router_info].append(port_id)
else:
in_ns_ports[router_info] = [port_id]
return list(in_ns_ports.items())
def _get_routers_in_project(self, context, project_id):
return self.core_plugin.get_routers(
context,
filters={'project_id': [project_id]})
def _router_dict_to_obj(self, r):
# The callbacks expect a router-info object
return router_info.RouterInfo(
None, r['id'], router=r,
agent_conf=None,
interface_driver=None,
use_ipv6=False)
def get_port_fwg(self, context, port_id):
"""Return the firewall group of this port
if the FWaaS rules should be added to the backend router.
"""
if not self.fwaas_enabled:
return False
ctx = context.elevated()
fwg_id = self._get_port_firewall_group_id(ctx, port_id)
if fwg_id is None:
# No FWaas Firewall was assigned to this port
return
# check the state of this firewall group
fwg = self._get_fw_group_from_plugin(ctx, fwg_id)
if fwg is not None:
if fwg.get('status') in (nl_constants.ERROR,
nl_constants.PENDING_DELETE):
# Do not add rules of firewalls with errors
LOG.warning("Port %(port)s will not get rules from firewall "
"group %(fwg)s which is in %(status)s",
{'port': port_id, 'fwg': fwg_id,
'status': fwg['status']})
return
return fwg
def _get_fw_group_from_plugin(self, context, fwg_id):
# NOTE(asarfaty): currently there is no api to get a specific firewall
fwg_list = self.fwplugin_rpc.get_firewall_groups_for_project(context)
for fwg in fwg_list:
if fwg['id'] == fwg_id:
return fwg
# TODO(asarfaty): add this api to fwaas firewall_db_v2
def _get_port_firewall_group_id(self, context, port_id):
entry = context.session.query(
firewall_db_v2.FirewallGroupPortAssociation).filter_by(
port_id=port_id).first()
if entry:
return entry.firewall_group_id

View File

@ -0,0 +1,29 @@
# 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_fwaas.common import fwaas_constants
from neutron_lib.plugins import directory
def is_fwaas_v1_plugin_enabled():
fwaas_plugin = directory.get_plugin(fwaas_constants.FIREWALL)
if fwaas_plugin:
return True
def is_fwaas_v2_plugin_enabled():
fwaas_plugin = directory.get_plugin(fwaas_constants.FIREWALL_V2)
if fwaas_plugin:
return True

View File

@ -13,171 +13,27 @@
# License for the specific language governing permissions and limitations
# under the License.
import netaddr
from neutron_fwaas.services.firewall.drivers import fwaas_base
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 import context as n_context
from neutron_lib.exceptions import firewall_v1 as exceptions
from neutron_lib.plugins import directory
from oslo_log import helpers as log_helpers
from oslo_log import log as logging
from vmware_nsx.db import db as nsx_db
from vmware_nsxlib.v3 import nsx_constants as consts
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 NSX-V3 driver'
RULE_NAME_PREFIX = 'Fwaas-'
DEFAULT_RULE_NAME = 'Default LR Layer3 Rule'
FWAAS_DRIVER_NAME = 'Fwaas V1 NSX-V3 driver'
NSX_FW_TAG = 'os-neutron-fw-id'
class EdgeFwaasV3Driver(fwaas_base.FwaasDriverBase):
class EdgeFwaasV3Driver(base_driver.CommonEdgeFwaasV3Driver):
"""NSX-V3 driver for Firewall As A Service - V1."""
def __init__(self):
LOG.debug("Loading FWaaS NsxV3Driver.")
super(EdgeFwaasV3Driver, self).__init__()
self.driver_name = FWAAS_DRIVER_NAME
self.backend_support = True
registry.subscribe(
self.check_backend_version,
resources.PROCESS, events.BEFORE_SPAWN)
@property
def nsxlib(self):
return directory.get_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, **kwargs):
if 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 should_apply_firewall_to_router(self, router_data):
"""Return True if the firewall rules should be added the router
Right now the driver supports for all routers.
"""
return True
@staticmethod
def _translate_action(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 exceptions.FirewallInternalDriverError(
driver=FWAAS_DRIVER_NAME)
def _translate_cidr(self, cidr):
return self.nsx_firewall.get_ip_cidr_reference(
cidr,
consts.IPV6 if netaddr.valid_ipv6(cidr) else consts.IPV4)
def _translate_addresses(self, cidrs):
return [self._translate_cidr(ip) for ip in 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):
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'):
nsx_rule['destinations'] = self._translate_addresses(
[rule['destination_ip_address']])
if rule.get('source_ip_address'):
nsx_rule['sources'] = self._translate_addresses(
[rule['source_ip_address']])
if rule.get('protocol'):
nsx_rule['services'] = self._translate_services(rule)
translated_rules.append(nsx_rule)
return translated_rules
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
@ -190,13 +46,6 @@ class EdgeFwaasV3Driver(fwaas_base.FwaasDriverBase):
self._update_backend_routers(context, apply_list, firewall['id'],
rules=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 exceptions.FirewallInternalDriverError(
driver=self.driver_name)
@log_helpers.log_method_call
def create_firewall(self, agent_mode, apply_list, firewall):
"""Create the Firewall with a given policy. """
@ -251,31 +100,6 @@ class EdgeFwaasV3Driver(fwaas_base.FwaasDriverBase):
self._update_nsx_router_firewall(context, router_id, fw_id,
rules)
def _get_backend_router_and_fw_section(self, context, router_id):
# find the backend router id in the DB
nsx_router_id = nsx_db.get_nsx_router_id(context.session, router_id)
if nsx_router_id is None:
LOG.error("Didn't find nsx router for router %s", router_id)
raise exceptions.FirewallInternalDriverError(
driver=self.driver_name)
# get the FW section id of the backend router
try:
section_id = self.nsx_router.get_firewall_section_id(
nsx_router_id)
except Exception as e:
LOG.error("Failed to find router firewall section for router "
"%(id)s: %(e)s", {'id': router_id, 'e': e})
raise exceptions.FirewallInternalDriverError(
driver=self.driver_name)
if section_id is None:
LOG.error("Failed to find router firewall section for router "
"%(id)s.", {'id': router_id})
raise exceptions.FirewallInternalDriverError(
driver=self.driver_name)
return nsx_router_id, section_id
def _update_nsx_router_tags(self, nsx_router_id, fw_id=None):
"""Get the updated tags to put on the nsx-router
@ -315,17 +139,11 @@ class EdgeFwaasV3Driver(fwaas_base.FwaasDriverBase):
"""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(
nsx_router_id, section_id = self.get_backend_router_and_fw_section(
context, router_id)
# Add default allow all rule
old_default_rule = self.nsx_firewall.get_default_rule(
section_id)
allow_all = {
'display_name': DEFAULT_RULE_NAME,
'action': consts.FW_ACTION_ALLOW,
'is_default': True,
'id': old_default_rule['id'] if old_default_rule else 0}
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])
@ -344,18 +162,12 @@ class EdgeFwaasV3Driver(fwaas_base.FwaasDriverBase):
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(
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
old_default_rule = self.nsx_firewall.get_default_rule(
section_id)
drop_all = {
'display_name': DEFAULT_RULE_NAME,
'action': consts.FW_ACTION_DROP,
'is_default': True,
'id': old_default_rule['id'] if old_default_rule else 0}
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])

View File

@ -0,0 +1,228 @@
# 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.
import netaddr
from neutron_fwaas.services.firewall.drivers import fwaas_base
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 oslo_log import log as logging
from vmware_nsx.db import db as nsx_db
from vmware_nsxlib.v3 import nsx_constants as consts
LOG = logging.getLogger(__name__)
RULE_NAME_PREFIX = 'Fwaas-'
DEFAULT_RULE_NAME = 'Default LR Layer3 Rule'
class CommonEdgeFwaasV3Driver(fwaas_base.FwaasDriverBase):
"""Base class for NSX-V3 driver for Firewall As A Service - V1 & V2."""
def __init__(self, driver_exception, driver_name):
super(CommonEdgeFwaasV3Driver, self).__init__()
self.driver_name = driver_name
self.backend_support = True
self.driver_exception = driver_exception
registry.subscribe(
self.check_backend_version,
resources.PROCESS, events.BEFORE_SPAWN)
@property
def core_plugin(self):
return directory.get_plugin()
@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, **kwargs):
if 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 should_apply_firewall_to_router(self, router_data):
"""Return True if the firewall rules should be added the router
Right now the driver supports for all routers.
"""
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):
return self.nsx_firewall.get_ip_cidr_reference(
cidr,
consts.IPV6 if netaddr.valid_ipv6(cidr) else consts.IPV4)
def _translate_addresses(self, cidrs):
return [self._translate_cidr(ip) for ip in 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):
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 replace_dest:
# set this value as the destination logical port,
# and set the rule to ingress
nsx_rule['destinations'] = [{'target_type': 'LogicalPort',
'target_id': replace_dest}]
nsx_rule['direction'] = 'IN'
elif rule.get('destination_ip_address'):
nsx_rule['destinations'] = self._translate_addresses(
[rule['destination_ip_address']])
if replace_src:
# set this value as the source logical port,
# and set the rule to eggress
nsx_rule['sources'] = [{'target_type': 'LogicalPort',
'target_id': replace_src}]
nsx_rule['direction'] = 'OUT'
elif rule.get('source_ip_address'):
nsx_rule['sources'] = self._translate_addresses(
[rule['source_ip_address']])
if rule.get('protocol'):
nsx_rule['services'] = self._translate_services(rule)
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_backend_router_and_fw_section(self, context, router_id):
# find the backend router id in the DB
nsx_router_id = nsx_db.get_nsx_router_id(context.session, router_id)
if nsx_router_id is None:
LOG.error("Didn't find nsx router for router %s", router_id)
raise self.driver_exception(driver=self.driver_name)
# get the FW section id of the backend router
try:
section_id = self.nsx_router.get_firewall_section_id(
nsx_router_id)
except Exception as e:
LOG.error("Failed to find router firewall section for router "
"%(id)s: %(e)s", {'id': router_id, 'e': e})
raise self.driver_exception(driver=self.driver_name)
if section_id is None:
LOG.error("Failed to find router firewall section for router "
"%(id)s.", {'id': router_id})
raise self.driver_exception(driver=self.driver_name)
return nsx_router_id, section_id
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}

View File

@ -0,0 +1,129 @@
# 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_v2 as exceptions
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'
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)
@log_helpers.log_method_call
def create_firewall_group(self, agent_mode, apply_list, firewall_group):
"""Create the Firewall with a given policy. """
self._validate_firewall_group(firewall_group)
self._update_backend_routers(apply_list, firewall_group['id'])
@log_helpers.log_method_call
def update_firewall_group(self, agent_mode, apply_list, firewall_group):
"""Remove previous policy and apply the new policy."""
self._validate_firewall_group(firewall_group)
self._update_backend_routers(apply_list, firewall_group['id'])
@log_helpers.log_method_call
def delete_firewall_group(self, agent_mode, apply_list, firewall_group):
"""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_group['id'])
@log_helpers.log_method_call
def apply_default_policy(self, agent_mode, apply_list, firewall_group):
"""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_group['id'])
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 unsupported routers
if not self.should_apply_firewall_to_router(router_info.router):
continue
routers.add(router_info.router_id)
# update each router once
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):
"""Return the list of translated rules per port"""
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(
firewall_group['ingress_rule_list'],
replace_dest=nsx_port_id))
port_rules.extend(self._translate_rules(
firewall_group['egress_rule_list'],
replace_src=nsx_port_id))
# Add ingress/egress block rules for this port
port_rules.extend([
{'display_name': "Block port ingress",
'action': consts.FW_ACTION_DROP,
'destinations': [{'target_type': 'LogicalPort',
'target_id': nsx_port_id}],
'direction': 'IN'},
{'display_name': "Block port egress",
'action': consts.FW_ACTION_DROP,
'sources': [{'target_type': 'LogicalPort',
'target_id': nsx_port_id}],
'direction': 'OUT'}])
return port_rules
def _validate_firewall_group(self, firewall_group):
"""Validate the rules in the firewall group"""
for rule in firewall_group['egress_rule_list']:
if rule.get('source_ip_address'):
# this rule cannot be used as egress rule
LOG.error("Rule %(id)s cannot be used in an egress "
"policy because it has a source",
{'id': rule['id']})
raise exceptions.FirewallInternalDriverError(
driver=FWAAS_DRIVER_NAME)
for rule in firewall_group['ingress_rule_list']:
if rule.get('destination_ip_address'):
# this rule cannot be used as ingress rule
LOG.error("Rule %(id)s cannot be used in an ingress "
"policy because it has a destination",
{'id': rule['id']})
raise exceptions.FirewallInternalDriverError(
driver=FWAAS_DRIVER_NAME)

View File

@ -0,0 +1,60 @@
# 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_v2 as \
com_callbacks
LOG = logging.getLogger(__name__)
class Nsxv3FwaasCallbacksV2(com_callbacks.NsxFwaasCallbacksV2):
"""NSX-V3 RPC callbacks for Firewall As A Service - V2."""
def __init__(self, nsxlib):
super(Nsxv3FwaasCallbacksV2, 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(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.fwaas_driver.should_apply_firewall_to_router(router_data):
return False
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_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(
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)

View File

@ -19,6 +19,7 @@ import mock
from vmware_nsxlib.v3 import nsx_constants as consts
from vmware_nsx.services.fwaas.nsx_v3 import edge_fwaas_driver
from vmware_nsx.services.fwaas.nsx_v3 import edge_fwaas_driver_base
from vmware_nsx.tests.unit.nsx_v3 import test_plugin as test_v3_plugin
FAKE_FW_ID = 'fake_fw_uuid'
@ -27,7 +28,7 @@ MOCK_NSX_ID = '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.DEFAULT_RULE_NAME,
'display_name': edge_fwaas_driver_base.DEFAULT_RULE_NAME,
'id': MOCK_DEFAULT_RULE_ID,
'action': consts.FW_ACTION_DROP}

View File

@ -0,0 +1,304 @@
# 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.
import copy
import mock
from neutron_lib.exceptions import firewall_v2 as exceptions
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
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_NSX_PORT_ID = 'fake_nsx_port_uuid'
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,
'id': MOCK_DEFAULT_RULE_ID,
'action': consts.FW_ACTION_DROP}
class Nsxv3FwaasTestCase(test_v3_plugin.NsxV3PluginTestCaseMixin):
def setUp(self):
super(Nsxv3FwaasTestCase, self).setUp()
self.firewall = edge_fwaas_driver_v2.EdgeFwaasV3DriverV2()
# Start some nsxlib/DB mocks
mock.patch(
"vmware_nsxlib.v3.core_resources.NsxLibLogicalRouter."
"get_firewall_section_id",
return_value=MOCK_SECTION_ID).start()
mock.patch(
"vmware_nsxlib.v3.security.NsxLibFirewallSection."
"get_default_rule",
return_value={'id': MOCK_DEFAULT_RULE_ID}).start()
mock.patch(
"vmware_nsx.db.db.get_nsx_router_id",
return_value=MOCK_NSX_ID).start()
self.plugin = directory.get_plugin()
self.plugin.fwaas_callbacks_v2 = fwaas_callbacks_v2.\
Nsxv3FwaasCallbacksV2(self.plugin.nsxlib)
self.plugin.fwaas_callbacks_v2.fwaas_enabled = True
self.plugin.fwaas_callbacks_v2.fwaas_driver = self.firewall
def _default_rule(self):
rule = DEFAULT_RULE
rule['action'] = consts.FW_ACTION_ALLOW
return rule
def _fake_rules_v4(self, is_ingress=True):
rule1 = {'enabled': True,
'action': 'allow',
'ip_version': 4,
'protocol': 'tcp',
'destination_port': '80',
'id': 'fake-fw-rule1',
'description': 'first rule'}
rule2 = {'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:
# source ips are allowed
rule1['source_ip_address'] = '10.24.4.2'
else:
# dest ips are allowed for egress rules
rule1['destination_ip_address'] = '10.24.4.2'
return [rule1, rule2, rule3, rule4]
def _fake_translated_rules(self, nsx_port_id, is_ingress=True):
# The expected translation of the rules in _fake_rules_v4
service1 = {'l4_protocol': 'TCP',
'resource_type': 'L4PortSetNSService',
'destination_ports': ['80'],
'source_ports': []}
rule1 = {'action': 'ALLOW',
'services': [{'service': service1}],
'sources': [{'target_id': '10.24.4.2',
'target_type': 'IPv4Address'}],
'display_name': 'Fwaas-fake-fw-rule1',
'notes': 'first rule'}
if not is_ingress:
rule1['destinations'] = rule1['sources']
del rule1['sources']
service2 = {'l4_protocol': 'TCP',
'resource_type': 'L4PortSetNSService',
'destination_ports': ['22-24'],
'source_ports': ['1-65535']}
rule2 = {'action': 'DROP', # Reject is replaced with deny
'services': [{'service': service2}],
'display_name': 'Fwaas-fake-fw-rule2'}
service3_1 = {'resource_type': 'ICMPTypeNSService',
'protocol': 'ICMPv4'}
service3_2 = {'resource_type': 'ICMPTypeNSService',
'protocol': 'ICMPv6'}
rule3 = {'action': 'DROP',
# icmp is translated to icmp v4 & v6
'services': [{'service': service3_1},
{'service': service3_2}],
'display_name': 'Fwaas-fake-fw-rule3'}
rule4 = {'action': 'DROP',
'display_name': 'Fwaas-fake-fw-rule4'}
if nsx_port_id:
if is_ingress:
field = 'destinations'
direction = 'IN'
else:
field = 'sources'
direction = 'OUT'
new_val = [{'target_id': nsx_port_id,
'target_type': 'LogicalPort'}]
for rule in (rule1, rule2, rule3, rule4):
rule[field] = new_val
rule['direction'] = direction
return [rule1, rule2, rule3, rule4]
def _fake_empty_firewall_group(self):
fw_inst = {'id': FAKE_FW_ID,
'admin_state_up': True,
'tenant_id': 'tenant-uuid',
'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': 'tenant-uuid',
'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}
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}
with mock.patch.object(self.plugin, '_get_router_interfaces',
return_value=[port]),\
mock.patch.object(self.plugin.fwaas_callbacks_v2, '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)),\
mock.patch("vmware_nsxlib.v3.security.NsxLibFirewallSection."
"update") as update_fw:
self.firewall.create_firewall_group('nsx', apply_list, firewall)
# expecting 2 block rules for the logical port (egress & ingress)
# and last default allow all rule
expected_rules = [
{'display_name': "Block port ingress",
'action': consts.FW_ACTION_DROP,
'destinations': [{'target_type': 'LogicalPort',
'target_id': FAKE_NSX_PORT_ID}],
'direction': 'IN'},
{'display_name': "Block port egress",
'action': consts.FW_ACTION_DROP,
'sources': [{'target_type': 'LogicalPort',
'target_id': FAKE_NSX_PORT_ID}],
'direction': 'OUT'},
self._default_rule()
]
update_fw.assert_called_once_with(
MOCK_SECTION_ID,
rules=expected_rules)
def _setup_firewall_with_rules(self, func, is_ingress=True):
apply_list = self._fake_apply_list()
rule_list = self._fake_rules_v4(is_ingress=is_ingress)
firewall = self._fake_firewall_group(rule_list, is_ingress=is_ingress)
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',
return_value=firewall),\
mock.patch("vmware_nsx.db.db.get_nsx_switch_and_port_id",
return_value=(0, FAKE_NSX_PORT_ID)),\
mock.patch("vmware_nsxlib.v3.security.NsxLibFirewallSection."
"update") as update_fw:
func('nsx', apply_list, firewall)
expected_rules = self._fake_translated_rules(
FAKE_NSX_PORT_ID, is_ingress=is_ingress) + [
{'display_name': "Block port ingress",
'action': consts.FW_ACTION_DROP,
'destinations': [{'target_type': 'LogicalPort',
'target_id': FAKE_NSX_PORT_ID}],
'direction': 'IN'},
{'display_name': "Block port egress",
'action': consts.FW_ACTION_DROP,
'sources': [{'target_type': 'LogicalPort',
'target_id': FAKE_NSX_PORT_ID}],
'direction': 'OUT'},
self._default_rule()
]
update_fw.assert_called_once_with(
MOCK_SECTION_ID,
rules=expected_rules)
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_illegal_rules(self):
"""Use ingress rules as the egress list and verify failure"""
apply_list = self._fake_apply_list()
rule_list = self._fake_rules_v4(is_ingress=True)
firewall = self._fake_firewall_group(rule_list, is_ingress=False)
self.assertRaises(exceptions.FirewallInternalDriverError,
self.firewall.create_firewall_group, 'nsx',
apply_list, firewall)
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_v2, '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)),\
mock.patch("vmware_nsxlib.v3.security.NsxLibFirewallSection."
"update") as update_fw:
self.firewall.delete_firewall_group('nsx', apply_list, firewall)
update_fw.assert_called_once_with(
MOCK_SECTION_ID,
rules=[self._default_rule()])
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("vmware_nsxlib.v3.security.NsxLibFirewallSection."
"update") as update_fw:
self.firewall.create_firewall_group('nsx', apply_list, firewall)
update_fw.assert_called_once_with(
MOCK_SECTION_ID,
rules=[self._default_rule()])

View File

@ -842,6 +842,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
def test_floatingip_create_different_fixed_ip_same_port(self):
self.skipTest('Multiple fixed ips on a port are not supported')