x/vmware-nsx
Code Issues Proposed changes

Merge "NSX|V: enable allow_address_pairs upon request"

Browse Source
This commit is contained in:
Zuul 2019-03-06 17:36:39 +00:00 committed by Gerrit Code Review
commit 9fbfaf22a7
3 changed files with 222 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
vmware_nsx/common/config.py
View File

@ -808,6 +808,10 @@ nsxv_opts = [
default=False,
help=_("Use subnet's exclusive router as a platform for "
"LBaaS")),
cfg.BoolOpt('allow_multiple_ip_addresses',
default=False,
help=_("Allow associating multiple IPs to VMs "
"without spoofguard limitations")),
]
# define the configuration of each NSX-V availability zone.

116
vmware_nsx/plugins/nsx_v/plugin.py
View File

@ -1355,8 +1355,15 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
net_data['id'])
net_data[psec.PORTSECURITY] = False
# Create SpoofGuard policy for network anti-spoofing
# allow_multiple_addresses will be overridden in case the user
# requires allowing multiple or cidr-based allowed address pairs
# defined per port but doesn't want to disable spoofguard globally
sg_policy_id = None
if cfg.CONF.nsxv.spoofguard_enabled and backend_network:
allow_multiple_addresses = (not net_data[psec.PORTSECURITY]
and cfg.CONF.nsxv.
allow_multiple_ip_addresses)
if (cfg.CONF.nsxv.spoofguard_enabled and backend_network and not
allow_multiple_addresses):
# This variable is set as the method below may result in a
# exception and we may need to rollback
predefined = False
@ -1742,6 +1749,75 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
"installed in order for security-groups to "
"function properly.", net_id)
def allow_multiple_addresses_configure_spoofguard(self, context, id,
net_attrs, net_morefs):
# User requires multiple addresses to be assigned to compute port
# and therefore, the spoofguard policy is being removed for this net.
orig_net = self.get_network(context, id)
if not net_attrs[psec.PORTSECURITY]:
sg_policy = nsxv_db.get_spoofguard_policy_id(context.session,
orig_net['id'])
if sg_policy:
try:
self.nsx_v.vcns.delete_spoofguard_policy(sg_policy)
except Exception as e:
LOG.error('Unable to delete spoofguard policy '
'%(sg_policy)s. Error: %(e)s',
{'sg_policy': sg_policy, 'e': e})
else:
LOG.warning("Could not locate spoofguard policy for "
"network %s", id)
# User requires port-security-enabled set to True and thus requires
# spoofguard installed for this network
else:
# Verifying that all ports are legal, i.e. not CIDR/subnet
filters = {'network_id': [id]}
ports = self.get_ports(context, filters=filters)
valid_ports = []
if ports:
for port in ports:
if self._is_compute_port(port):
for ap in port[addr_apidef.ADDRESS_PAIRS]:
if len(ap['ip_address'].split('/')) > 1:
msg = _('Port %s contains CIDR/subnet, '
'which is not supported at the '
'backend ') % port['id']
raise n_exc.BadRequest(
resource='ports',
msg=msg)
else:
valid_ports.append(port)
try:
sg_policy_id, predefined = (
self._prepare_spoofguard_policy(
orig_net.get(pnet.NETWORK_TYPE), orig_net,
net_morefs))
if sg_policy_id:
nsxv_db.map_spoofguard_policy_for_network(
context.session,
orig_net['id'], sg_policy_id)
except Exception as e:
msg = _('Unable to create spoofguard policy, error: %('
'error)s, '
'net_morefs=%(net_morefs)s, network_id= %('
'network_id)s') % {'error': e, 'net_morefs':
net_morefs, 'network_id': orig_net}
raise n_exc.BadRequest(resource='spoofguard policy', msg=msg)
try:
for port in valid_ports:
vnic_idx = port.get(ext_vnic_idx.VNIC_INDEX)
device_id = port['device_id']
vnic_id = self._get_port_vnic_id(vnic_idx,
device_id)
self._update_vnic_assigned_addresses(context.session, port,
vnic_id)
except Exception as e:
msg = _('Unable to add port to spoofguard policy error '
'%s') % e
raise n_exc.BadRequest(resource='spoofguard policy',
msg=msg)
def update_network(self, context, id, network):
net_attrs = network['network']
orig_net = self.get_network(context, id)
@ -1767,6 +1843,15 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
if psec_update:
self._update_network_validate_port_sec(context, id, net_attrs)
# Change spoofguard accordingly - either remove if
# port-security-enabled was set to false or add (with relevant ports)
# if set to true.
if (cfg.CONF.nsxv.spoofguard_enabled and
cfg.CONF.nsxv.allow_multiple_ip_addresses and psec_update):
self.allow_multiple_addresses_configure_spoofguard(context, id,
net_attrs,
net_morefs)
# Check if the physical network of a vlan provider network was updated
updated_morefs = False
if (net_attrs.get(pnet.PHYSICAL_NETWORK) and
@ -1875,17 +1960,24 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
return net_res
def _validate_address_pairs(self, attrs, db_port):
def _validate_address_pairs(self, context, attrs, db_port):
network_port_security = self._get_network_security_binding(
context, db_port['network_id'])
if (not cfg.CONF.nsxv.allow_multiple_ip_addresses and
not network_port_security):
for ap in attrs[addr_apidef.ADDRESS_PAIRS]:
# Check that the IP address is a subnet
if len(ap['ip_address'].split('/')) > 1:
msg = _('NSXv does not support CIDR as address pairs')
raise n_exc.BadRequest(resource='address_pairs',
msg=msg)
# Check that the MAC address is the same as the port
for ap in attrs[addr_apidef.ADDRESS_PAIRS]:
# Check that the IP address is a subnet
if len(ap['ip_address'].split('/')) > 1:
msg = _('NSXv does not support CIDR as address pairs')
raise n_exc.BadRequest(resource='address_pairs', msg=msg)
# Check that the MAC address is the same as the port
if ('mac_address' in ap and
ap['mac_address'] != db_port['mac_address']):
msg = _('Address pairs should have same MAC as the port')
raise n_exc.BadRequest(resource='address_pairs', msg=msg)
ap['mac_address'] != db_port['mac_address']):
msg = _('Address pairs should have same MAC as the '
'port')
raise n_exc.BadRequest(resource='address_pairs', msg=msg)
def _is_mac_in_use(self, context, network_id, mac_address):
# Override this method as the backed doesn't support using the same
@ -2004,7 +2096,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
if self._check_update_has_allowed_address_pairs(port):
if not port_security:
raise addr_exc.AddressPairAndPortSecurityRequired()
self._validate_address_pairs(attrs, neutron_db)
self._validate_address_pairs(context, attrs, neutron_db)
else:
# remove ATTR_NOT_SPECIFIED
attrs[addr_apidef.ADDRESS_PAIRS] = []
@ -2216,7 +2308,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
self._validate_extra_dhcp_options(dhcp_opts)
self._validate_port_qos(port_data)
if addr_apidef.ADDRESS_PAIRS in attrs:
self._validate_address_pairs(attrs, original_port)
self._validate_address_pairs(context, attrs, original_port)
self._validate_max_ips_per_port(
port_data.get('fixed_ips', []),
port_data.get('device_owner', original_port['device_owner']))

114
vmware_nsx/tests/unit/nsx_v/test_plugin.py
View File

@ -615,6 +615,33 @@ class TestNetworksV2(test_plugin.TestNetworksV2, NsxVPluginV2TestCase):
# Assert neutron name is not truncated
self.assertEqual(net['network']['name'], name)
def test_create_update_network_allow_multiple_addresses_spoofguard(self):
# allow_multiple_addresses flag is True, first step is to check that
# when port-security-allowed is false - spoofguard policy is not
# created.
# next step is to update port-security-allowed to true - spoofguard
# policy is now created for this network.
q_context = context.Context('', 'tenant_1')
providernet_args = {psec.PORTSECURITY: False}
cfg.CONF.set_default('allow_multiple_ip_addresses', True, 'nsxv')
res = self._create_network(fmt='json', name='net-1',
admin_state_up=True,
providernet_args=providernet_args,
arg_list=(psec.PORTSECURITY,))
network1 = self.deserialize(self.fmt, res)
net1_id = network1['network']['id']
# not creating spoofguard policy
self.assertIsNone(nsxv_db.get_spoofguard_policy_id(q_context.session,
net1_id))
args = {'network': {psec.PORTSECURITY: True}}
req = self.new_update_request('networks', args,
network1['network']['id'], fmt='json')
res = self.deserialize('json', req.get_response(self.api))
net1_id = res['network']['id']
# creating spoofguard policy
self.assertIsNotNone(nsxv_db.get_spoofguard_policy_id(
q_context.session, net1_id))
def test_update_network_with_admin_false(self):
data = {'network': {'admin_state_up': False}}
with self.network() as net:
@ -1769,6 +1796,93 @@ class TestPortsV2(NsxVPluginV2TestCase,
self.assertEqual("PortSecurityAndIPRequiredForSecurityGroups",
res['NeutronError']['type'])
def test_port_add_to_spoofguard_allow_multiple_addresses(self):
# allow_multiple_addresses flag is True, first step is to check that
# when port-security-allowed is false - spoofguard policy is not
# created.
# next step is to update port-security-allowed to true - spoofguard
# policy is now created for this network.
providernet_args = {psec.PORTSECURITY: False}
cfg.CONF.set_default('allow_multiple_ip_addresses', True, 'nsxv')
res = self._create_network(fmt='json', name='net-1',
admin_state_up=True,
providernet_args=providernet_args,
arg_list=(psec.PORTSECURITY,))
network1 = self.deserialize(self.fmt, res)
net1_id = network1['network']['id']
with self.subnet(network=network1, cidr='10.0.0.0/24'):
# create a compute port with port security
address_pairs = [{'ip_address': '192.168.1.1'}]
device_id = _uuid()
vnic_index = 3
compute_port_create = self._create_port(
'json', net1_id,
arg_list=(
'port_security_enabled',
'device_id',
'device_owner',
'allowed_address_pairs',),
port_security_enabled=True,
device_id=device_id,
device_owner='compute:None',
allowed_address_pairs=address_pairs)
port = self.deserialize('json', compute_port_create)
port = self._update_port_index(
port['port']['id'], device_id, vnic_index)
# Verify the port is added to the spoofguard policy
with mock.patch.object(
self.plugin, '_update_vnic_assigned_addresses') as \
update_approved_port:
args = {'network': {psec.PORTSECURITY: True}}
req = self.new_update_request('networks', args, net1_id,
fmt='json')
req.get_response(self.api)
# The expected vnic-id format by NsxV
update_approved_port.assert_called_once_with(
mock.ANY, mock.ANY, '%s.%03d' % (device_id, vnic_index))
def test_port_add_to_spoofguard_allow_multiple_addresses_fail(self):
# allow_multiple_addresses flag is True, first step is to check that
# when port-security-allowed is false - spoofguard policy is not
# created.
# next step is to update port-security-allowed to true but the port
# has CIDR defined as a address pair - action is aborted.
# policy is now created for this network.
providernet_args = {psec.PORTSECURITY: False}
cfg.CONF.set_default('allow_multiple_ip_addresses', True, 'nsxv')
res = self._create_network(fmt='json', name='net-1',
admin_state_up=True,
providernet_args=providernet_args,
arg_list=(psec.PORTSECURITY,))
network1 = self.deserialize(self.fmt, res)
net1_id = network1['network']['id']
with self.subnet(network=network1, cidr='10.0.0.0/24'):
# create a compute port with port security
address_pairs = [{'ip_address': '192.168.1.0/24'}]
device_id = _uuid()
vnic_index = 3
compute_port_create = self._create_port(
'json', net1_id,
arg_list=(
'port_security_enabled',
'device_id',
'device_owner',
'allowed_address_pairs',),
port_security_enabled=True,
device_id=device_id,
device_owner='compute:None',
allowed_address_pairs=address_pairs)
port = self.deserialize('json', compute_port_create)
port = self._update_port_index(
port['port']['id'], device_id, vnic_index)
# Action is failed due to CIDR defined in the port.
args = {'network': {psec.PORTSECURITY: True}}
plugin = directory.get_plugin()
self.assertRaises(n_exc.BadRequest,
plugin.update_network,
context.get_admin_context(),
net1_id, args)
class TestSubnetsV2(NsxVPluginV2TestCase,
test_plugin.TestSubnetsV2):