openstack
/
neutron
Code Issues Proposed changes

Merge "Check whether vxlan group and local addresses are IPv4 or IPv6"

This commit is contained in:
Zuul 2022-01-11 09:36:16 +00:00 committed by Gerrit Code Review
parent 051face9d7 a47e9494c1
commit d1b5efe5ae
3 changed files with 114 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
neutron/agent/linux/ip_lib.py
View File

@ -283,13 +283,31 @@ class IPWrapper(SubProcessBase):
kwargs = {'vxlan_id': vni,
'physical_interface': dev}
if group:
kwargs['vxlan_group'] = group
try:
ip_version = common_utils.get_ip_version(group)
if ip_version == constants.IP_VERSION_6:
kwargs['vxlan_group6'] = group
else:
kwargs['vxlan_group'] = group
except netaddr.core.AddrFormatError:
err_msg = _("Invalid group address: %s") % group
raise exceptions.InvalidInput(error_message=err_msg)
if dev:
kwargs['physical_interface'] = dev
if ttl:
kwargs['vxlan_ttl'] = ttl
if tos:
kwargs['vxlan_tos'] = tos
if local:
kwargs['vxlan_local'] = local
try:
ip_version = common_utils.get_ip_version(local)
if ip_version == constants.IP_VERSION_6:
kwargs['vxlan_local6'] = local
else:
kwargs['vxlan_local'] = local
except netaddr.core.AddrFormatError:
err_msg = _("Invalid local address: %s") % local
raise exceptions.InvalidInput(error_message=err_msg)
if proxy:
kwargs['vxlan_proxy'] = proxy
# tuple: min,max

17
neutron/tests/functional/agent/linux/test_ip_lib.py
View File

@ -49,6 +49,7 @@ TEST_IP_NEIGH = '240.0.0.2'
TEST_IP_SECONDARY = '240.0.0.3'
TEST_IP6_NEIGH = 'fd00::2'
TEST_IP6_SECONDARY = 'fd00::3'
TEST_IP6_VXLAN_GROUP = 'ff00::1'
TEST_IP_NUD_STATES = ((TEST_IP_NEIGH, 'permanent'),
(TEST_IP_SECONDARY, 'reachable'),
(TEST_IP6_NEIGH, 'permanent'),
@ -247,6 +248,22 @@ class IpLibTestCase(IpLibTestFramework):
device.link.delete()
self.assertFalse(ip_lib.vxlan_in_use(9999, namespace=attr.namespace))
def test_ipv6_vxlan_exists(self):
attr = self.generate_device_details(
name='test_device', ip_cidrs=["%s/24" % TEST_IP, 'fd00::1/64']
)
self.manage_device(attr)
ip = ip_lib.IPWrapper(namespace=attr.namespace)
ip.netns.add(attr.namespace)
self.addCleanup(ip.netns.delete, attr.namespace)
self.assertFalse(ip_lib.vxlan_in_use(9999, namespace=attr.namespace))
device = ip.add_vxlan('test_vxlan_device', 9999, local='fd00::1',
group=TEST_IP6_VXLAN_GROUP, dev='test_device')
self.addCleanup(self._safe_delete_device, device)
self.assertTrue(ip_lib.vxlan_in_use(9999, namespace=attr.namespace))
device.link.delete()
self.assertFalse(ip_lib.vxlan_in_use(9999, namespace=attr.namespace))
def test_ipwrapper_get_device_by_ip_None(self):
ip_wrapper = ip_lib.IPWrapper(namespace=None)
self.assertIsNone(ip_wrapper.get_device_by_ip(ip=None))

97
neutron/tests/unit/agent/linux/test_ip_lib.py
View File

@ -98,6 +98,16 @@ SUBNET_SAMPLE1 = ("10.0.0.0/24 dev qr-23380d11-d2 scope link src 10.0.0.1\n"
SUBNET_SAMPLE2 = ("10.0.0.0/24 dev tap1d7888a7-10 scope link src 10.0.0.2\n"
"10.0.0.0/24 dev qr-23380d11-d2 scope link src 10.0.0.1")
VXLAN4_GROUP_SAMPLE = "239.0.0.1"
VXLAN4_LOCAL_SAMPLE = "192.168.45.100"
VXLAN6_GROUP_SAMPLE = "ff00::1"
VXLAN6_INVALID_IPV6_SAMPLE = "invalid:ipv6::address"
VXLAN6_LOCAL_SAMPLE = "fd00::1"
class TestSubProcessBase(base.BaseTestCase):
def setUp(self):
@ -374,20 +384,54 @@ class TestIpWrapper(base.BaseTestCase):
'namespace': None,
'kind': 'vxlan',
'vxlan_id': 'vni0',
'vxlan_group': 'group0',
'vxlan_group': VXLAN4_GROUP_SAMPLE,
'physical_interface': 'dev0',
'vxlan_ttl': 'ttl0',
'vxlan_tos': 'tos0',
'vxlan_local': 'local0',
'vxlan_local': VXLAN4_LOCAL_SAMPLE,
'vxlan_proxy': True,
'vxlan_port_range': ('1', '2')}
retval = ip_lib.IPWrapper().add_vxlan('vxlan0', 'vni0',
group=VXLAN4_GROUP_SAMPLE,
dev='dev0', ttl='ttl0',
tos='tos0',
local=VXLAN4_LOCAL_SAMPLE,
proxy=True, srcport=(1, 2))
self.assertIsInstance(retval, ip_lib.IPDevice)
self.assertEqual(retval.name, 'vxlan0')
self.assertDictEqual(expected_call_params, self.call_params)
@mock.patch.object(priv_lib, 'create_interface')
def test_add_vxlan6_valid_srcport_length(self, create):
self.call_params = {}
def fake_create_interface(ifname, namespace, kind, **kwargs):
self.call_params = dict(
ifname=ifname,
namespace=namespace,
kind=kind,
**kwargs)
create.side_effect = fake_create_interface
expected_call_params = {
'ifname': 'vxlan0',
'namespace': None,
'kind': 'vxlan',
'vxlan_id': 'vni0',
'vxlan_group6': VXLAN6_GROUP_SAMPLE,
'physical_interface': 'dev0',
'vxlan_ttl': 'ttl0',
'vxlan_tos': 'tos0',
'vxlan_local6': VXLAN6_LOCAL_SAMPLE,
'vxlan_proxy': True,
'vxlan_port_range': ('1', '2')}
retval = ip_lib.IPWrapper().add_vxlan('vxlan0', 'vni0', 'dev0',
group='group0',
ttl='ttl0',
tos='tos0',
local='local0', proxy=True,
srcport=(1, 2))
group=VXLAN6_GROUP_SAMPLE,
ttl='ttl0', tos='tos0',
local=VXLAN6_LOCAL_SAMPLE,
proxy=True, srcport=(1, 2))
self.assertIsInstance(retval, ip_lib.IPDevice)
self.assertEqual(retval.name, 'vxlan0')
self.assertDictEqual(expected_call_params, self.call_params)
@ -395,19 +439,33 @@ class TestIpWrapper(base.BaseTestCase):
def test_add_vxlan_invalid_srcport_length(self):
wrapper = ip_lib.IPWrapper()
self.assertRaises(exceptions.NetworkVxlanPortRangeError,
wrapper.add_vxlan, 'vxlan0', 'vni0', group='group0',
dev='dev0', ttl='ttl0', tos='tos0',
local='local0', proxy=True,
wrapper.add_vxlan, 'vxlan0', 'vni0',
group=VXLAN4_GROUP_SAMPLE, dev='dev0', ttl='ttl0',
tos='tos0', local=VXLAN4_LOCAL_SAMPLE, proxy=True,
srcport=('1', '2', '3'))
def test_add_vxlan_invalid_srcport_range(self):
wrapper = ip_lib.IPWrapper()
self.assertRaises(exceptions.NetworkVxlanPortRangeError,
wrapper.add_vxlan, 'vxlan0', 'vni0', group='group0',
dev='dev0', ttl='ttl0', tos='tos0',
local='local0', proxy=True,
wrapper.add_vxlan, 'vxlan0', 'vni0',
group=VXLAN4_GROUP_SAMPLE, dev='dev0', ttl='ttl0',
tos='tos0', local=VXLAN4_LOCAL_SAMPLE, proxy=True,
srcport=(2000, 1000))
def test_add_vxlan_invalid_ipv6_groupaddr(self):
wrapper = ip_lib.IPWrapper()
self.assertRaises(exceptions.InvalidInput,
wrapper.add_vxlan, 'vxlan0', 'vni0',
group=VXLAN6_INVALID_IPV6_SAMPLE,
dev='dev0', ttl='ttl0', tos='tos0')
def test_add_vxlan_invalid_ipv6_localaddr(self):
wrapper = ip_lib.IPWrapper()
self.assertRaises(exceptions.InvalidInput,
wrapper.add_vxlan, 'vxlan0', 'vni0',
local=VXLAN6_INVALID_IPV6_SAMPLE, dev='dev0',
ttl='ttl0', tos='tos0')
@mock.patch.object(priv_lib, 'create_interface')
def test_add_vxlan_dstport(self, create):
self.call_params = {}
@ -425,21 +483,20 @@ class TestIpWrapper(base.BaseTestCase):
'namespace': None,
'kind': 'vxlan',
'vxlan_id': 'vni0',
'vxlan_group': 'group0',
'vxlan_group': VXLAN4_GROUP_SAMPLE,
'physical_interface': 'dev0',
'vxlan_ttl': 'ttl0',
'vxlan_tos': 'tos0',
'vxlan_local': 'local0',
'vxlan_local': VXLAN4_LOCAL_SAMPLE,
'vxlan_proxy': True,
'vxlan_port_range': ('1', '2'),
'vxlan_port': 4789}
retval = ip_lib.IPWrapper().add_vxlan('vxlan0', 'vni0', 'dev0',
group='group0',
ttl='ttl0',
tos='tos0',
local='local0', proxy=True,
srcport=(1, 2),
group=VXLAN4_GROUP_SAMPLE,
ttl='ttl0', tos='tos0',
local=VXLAN4_LOCAL_SAMPLE,
proxy=True, srcport=(1, 2),
dstport=4789)
self.assertIsInstance(retval, ip_lib.IPDevice)