Merge "Utility API to generate EUI-64 IPv6 address"

2 files changed, 125 insertions, 0 deletions

diff --git a/oslo_utils/netutils.py b/oslo_utils/netutils.py
index 41587f2..bdb4632 100644
--- a/oslo_utils/netutils.py
+++ b/oslo_utils/netutils.py
@@ -18,16 +18,19 @@ Network-related utilities and helper functions.
 """
 
 import logging
+import os
 import socket
 
 import netaddr
 import netifaces
 from six.moves.urllib import parse
 
+from oslo_utils._i18n import _
 from oslo_utils._i18n import _LI
 from oslo_utils._i18n import _LW
 
 LOG = logging.getLogger(__name__)
+_IS_IPV6_ENABLED = None
 
 
 def parse_host_port(address, default_port=None):
@@ -104,6 +107,57 @@ def is_valid_ipv6(address):
         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):
     """Verify that address represents a valid IP address.
 
diff --git a/oslo_utils/tests/test_netutils.py b/oslo_utils/tests/test_netutils.py
index 7e1b545..1bc4990 100644
--- a/oslo_utils/tests/test_netutils.py
+++ b/oslo_utils/tests/test_netutils.py
@@ -231,3 +231,74 @@ class NetworkUtilsTest(test_base.BaseTestCase):
         addr = netutils._get_my_ipv4_address()
         self.assertEqual('127.0.0.1', addr)
         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()