Utility API to generate EUI-64 IPv6 address
For a network using IPv6 SLAAC, Neutron would calculate the port addresses based on EUI-64 specification which is generated via IPv6 prefix and the interface MAC address. EUI-64 is a standard algorithm and is explained at the following url http://packetlife.net/blog/2008/aug/4/eui-64-ipv6 This utility api is present in both Neutron [1] and Tempest [2] to validate certain IPv6 scenarios. Hence, having it in oslo.utils would be ideal. [1] - https://github.com/openstack/neutron/blob/master/neutron/common/ipv6_utils.py#L32 [2] - https://github.com/openstack/tempest/blob/master/tempest/common/utils/data_utils.py#L85 Change-Id: I3d1c6b22ecddf7faad83cc12c674a9c5c96b1759
This commit is contained in:
parent
8a5a4d87da
commit
9d9818bf48
@ -18,16 +18,19 @@ Network-related utilities and helper functions.
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
import os
|
||||||
import socket
|
import socket
|
||||||
|
|
||||||
import netaddr
|
import netaddr
|
||||||
import netifaces
|
import netifaces
|
||||||
from six.moves.urllib import parse
|
from six.moves.urllib import parse
|
||||||
|
|
||||||
|
from oslo_utils._i18n import _
|
||||||
from oslo_utils._i18n import _LI
|
from oslo_utils._i18n import _LI
|
||||||
from oslo_utils._i18n import _LW
|
from oslo_utils._i18n import _LW
|
||||||
|
|
||||||
LOG = logging.getLogger(__name__)
|
LOG = logging.getLogger(__name__)
|
||||||
|
_IS_IPV6_ENABLED = None
|
||||||
|
|
||||||
|
|
||||||
def parse_host_port(address, default_port=None):
|
def parse_host_port(address, default_port=None):
|
||||||
@ -104,6 +107,57 @@ def is_valid_ipv6(address):
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def get_ipv6_addr_by_EUI64(prefix, mac):
|
||||||
|
"""Calculate IPv6 address using EUI-64 specification.
|
||||||
|
|
||||||
|
This method calculates the IPv6 address using the EUI-64
|
||||||
|
addressing scheme as explained in rfc2373.
|
||||||
|
|
||||||
|
:param prefix: IPv6 prefix.
|
||||||
|
:param mac: IEEE 802 48-bit MAC address.
|
||||||
|
:returns: IPv6 address on success.
|
||||||
|
:raises ValueError, TypeError: For any invalid input.
|
||||||
|
"""
|
||||||
|
# Check if the prefix is an IPv4 address
|
||||||
|
if netaddr.valid_ipv4(prefix):
|
||||||
|
msg = _("Unable to generate IP address by EUI64 for IPv4 prefix")
|
||||||
|
raise ValueError(msg)
|
||||||
|
try:
|
||||||
|
eui64 = int(netaddr.EUI(mac).eui64())
|
||||||
|
prefix = netaddr.IPNetwork(prefix)
|
||||||
|
return netaddr.IPAddress(prefix.first + eui64 ^ (1 << 57))
|
||||||
|
except (ValueError, netaddr.AddrFormatError):
|
||||||
|
raise ValueError(_('Bad prefix or mac format for generating IPv6 '
|
||||||
|
'address by EUI-64: %(prefix)s, %(mac)s:')
|
||||||
|
% {'prefix': prefix, 'mac': mac})
|
||||||
|
except TypeError:
|
||||||
|
raise TypeError(_('Bad prefix type for generating IPv6 address by '
|
||||||
|
'EUI-64: %s') % prefix)
|
||||||
|
|
||||||
|
|
||||||
|
def is_ipv6_enabled():
|
||||||
|
"""Check if IPv6 support is enabled on the platform.
|
||||||
|
|
||||||
|
This api will look into the proc entries of the platform to figure
|
||||||
|
out the status of IPv6 support on the platform.
|
||||||
|
|
||||||
|
:returns: True if the platform has IPv6 support, False otherwise.
|
||||||
|
|
||||||
|
"""
|
||||||
|
|
||||||
|
global _IS_IPV6_ENABLED
|
||||||
|
|
||||||
|
if _IS_IPV6_ENABLED is None:
|
||||||
|
disabled_ipv6_path = "/proc/sys/net/ipv6/conf/default/disable_ipv6"
|
||||||
|
if os.path.exists(disabled_ipv6_path):
|
||||||
|
with open(disabled_ipv6_path, 'r') as f:
|
||||||
|
disabled = f.read().strip()
|
||||||
|
_IS_IPV6_ENABLED = disabled == "0"
|
||||||
|
else:
|
||||||
|
_IS_IPV6_ENABLED = False
|
||||||
|
return _IS_IPV6_ENABLED
|
||||||
|
|
||||||
|
|
||||||
def is_valid_ip(address):
|
def is_valid_ip(address):
|
||||||
"""Verify that address represents a valid IP address.
|
"""Verify that address represents a valid IP address.
|
||||||
|
|
||||||
|
@ -231,3 +231,74 @@ class NetworkUtilsTest(test_base.BaseTestCase):
|
|||||||
addr = netutils._get_my_ipv4_address()
|
addr = netutils._get_my_ipv4_address()
|
||||||
self.assertEqual('127.0.0.1', addr)
|
self.assertEqual('127.0.0.1', addr)
|
||||||
self.assertFalse(ifaddr.called)
|
self.assertFalse(ifaddr.called)
|
||||||
|
|
||||||
|
|
||||||
|
class IPv6byEUI64TestCase(test_base.BaseTestCase):
|
||||||
|
"""Unit tests to generate IPv6 by EUI-64 operations."""
|
||||||
|
|
||||||
|
def test_generate_IPv6_by_EUI64(self):
|
||||||
|
addr = netutils.get_ipv6_addr_by_EUI64('2001:db8::',
|
||||||
|
'00:16:3e:33:44:55')
|
||||||
|
self.assertEqual('2001:db8::216:3eff:fe33:4455', addr.format())
|
||||||
|
|
||||||
|
def test_generate_IPv6_with_IPv4_prefix(self):
|
||||||
|
ipv4_prefix = '10.0.8'
|
||||||
|
mac = '00:16:3e:33:44:55'
|
||||||
|
self.assertRaises(ValueError, lambda:
|
||||||
|
netutils.get_ipv6_addr_by_EUI64(ipv4_prefix, mac))
|
||||||
|
|
||||||
|
def test_generate_IPv6_with_bad_mac(self):
|
||||||
|
bad_mac = '00:16:3e:33:44:5Z'
|
||||||
|
prefix = '2001:db8::'
|
||||||
|
self.assertRaises(ValueError, lambda:
|
||||||
|
netutils.get_ipv6_addr_by_EUI64(prefix, bad_mac))
|
||||||
|
|
||||||
|
def test_generate_IPv6_with_bad_prefix(self):
|
||||||
|
mac = '00:16:3e:33:44:55'
|
||||||
|
bad_prefix = 'bb'
|
||||||
|
self.assertRaises(ValueError, lambda:
|
||||||
|
netutils.get_ipv6_addr_by_EUI64(bad_prefix, mac))
|
||||||
|
|
||||||
|
def test_generate_IPv6_with_error_prefix_type(self):
|
||||||
|
mac = '00:16:3e:33:44:55'
|
||||||
|
prefix = 123
|
||||||
|
self.assertRaises(TypeError, lambda:
|
||||||
|
netutils.get_ipv6_addr_by_EUI64(prefix, mac))
|
||||||
|
|
||||||
|
|
||||||
|
class TestIsIPv6Enabled(test_base.BaseTestCase):
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
super(TestIsIPv6Enabled, self).setUp()
|
||||||
|
|
||||||
|
def reset_detection_flag():
|
||||||
|
netutils._IS_IPV6_ENABLED = None
|
||||||
|
reset_detection_flag()
|
||||||
|
self.addCleanup(reset_detection_flag)
|
||||||
|
self.mock_exists = mock.patch("os.path.exists",
|
||||||
|
return_value=True).start()
|
||||||
|
mock_open = mock.patch("six.moves.builtins.open").start()
|
||||||
|
self.mock_read = mock_open.return_value.__enter__.return_value.read
|
||||||
|
|
||||||
|
def test_enabled(self):
|
||||||
|
self.mock_read.return_value = "0"
|
||||||
|
enabled = netutils.is_ipv6_enabled()
|
||||||
|
self.assertTrue(enabled)
|
||||||
|
|
||||||
|
def test_disabled(self):
|
||||||
|
self.mock_read.return_value = "1"
|
||||||
|
enabled = netutils.is_ipv6_enabled()
|
||||||
|
self.assertFalse(enabled)
|
||||||
|
|
||||||
|
def test_disabled_non_exists(self):
|
||||||
|
self.mock_exists.return_value = False
|
||||||
|
enabled = netutils.is_ipv6_enabled()
|
||||||
|
self.assertFalse(enabled)
|
||||||
|
self.assertFalse(self.mock_read.called)
|
||||||
|
|
||||||
|
def test_memoize(self):
|
||||||
|
self.mock_read.return_value = "0"
|
||||||
|
netutils.is_ipv6_enabled()
|
||||||
|
enabled = netutils.is_ipv6_enabled()
|
||||||
|
self.assertTrue(enabled)
|
||||||
|
self.mock_read.assert_called_once_with()
|
||||||
|
Loading…
x
Reference in New Issue
Block a user