support network configuration in cloud-init --local

this allows 'cloud-init --local' to fully run before networking comes up.
By doing so, we can now cleanly apply networking to the new system.

This adds support for reading ConfigDrive network configuration
and also from NoCloud.  The support is only present for ubuntu/debian
at the current time.  Other distros will follow.

Also ability to specify network configuration on kernel command line
via either ip= or network-config=<base64>.

1 files changed, 127 insertions, 0 deletions

diff --git a/tests/unittests/test_net.py b/tests/unittests/test_net.py
new file mode 100644
index 00000000..dfb31710
--- /dev/null
+++ b/tests/unittests/test_net.py
@@ -0,0 +1,127 @@
+from cloudinit import util
+from cloudinit import net
+from .helpers import TestCase
+
+import base64
+import copy
+import io
+import gzip
+import json
+import os
+
+DHCP_CONTENT_1 = """
+DEVICE='eth0'
+PROTO='dhcp'
+IPV4ADDR='192.168.122.89'
+IPV4BROADCAST='192.168.122.255'
+IPV4NETMASK='255.255.255.0'
+IPV4GATEWAY='192.168.122.1'
+IPV4DNS0='192.168.122.1'
+IPV4DNS1='0.0.0.0'
+HOSTNAME='foohost'
+DNSDOMAIN=''
+NISDOMAIN=''
+ROOTSERVER='192.168.122.1'
+ROOTPATH=''
+filename=''
+UPTIME='21'
+DHCPLEASETIME='3600'
+DOMAINSEARCH='foo.com'
+"""
+
+DHCP_EXPECTED_1 = {
+    'name': 'eth0',
+    'type': 'physical',
+    'subnets': [{'broadcast': '192.168.122.255',
+                 'gateway': '192.168.122.1',
+                 'dns_search': ['foo.com'],
+                 'type': 'dhcp',
+                 'netmask': '255.255.255.0',
+                 'dns_nameservers': ['192.168.122.1']}],
+}
+
+
+STATIC_CONTENT_1 = """
+DEVICE='eth1'
+PROTO='static'
+IPV4ADDR='10.0.0.2'
+IPV4BROADCAST='10.0.0.255'
+IPV4NETMASK='255.255.255.0'
+IPV4GATEWAY='10.0.0.1'
+IPV4DNS0='10.0.1.1'
+IPV4DNS1='0.0.0.0'
+HOSTNAME='foohost'
+UPTIME='21'
+DHCPLEASETIME='3600'
+DOMAINSEARCH='foo.com'
+"""
+
+STATIC_EXPECTED_1 = {
+    'name': 'eth1',
+    'type': 'physical',
+    'subnets': [{'broadcast': '10.0.0.255', 'gateway': '10.0.0.1',
+                 'dns_search': ['foo.com'], 'type': 'static',
+                 'netmask': '255.255.255.0',
+                 'dns_nameservers': ['10.0.1.1']}],
+}
+
+
+class TestNetConfigParsing(TestCase):
+    simple_cfg = {
+        'config': [{"type": "physical", "name": "eth0",
+                    "mac_address": "c0:d6:9f:2c:e8:80",
+                    "subnets": [{"type": "dhcp"}]}]}
+
+    def test_klibc_convert_dhcp(self):
+        found = net._klibc_to_config_entry(DHCP_CONTENT_1)
+        self.assertEqual(found, ('eth0', DHCP_EXPECTED_1))
+
+    def test_klibc_convert_static(self):
+        found = net._klibc_to_config_entry(STATIC_CONTENT_1)
+        self.assertEqual(found, ('eth1', STATIC_EXPECTED_1))
+
+    def test_config_from_klibc_net_cfg(self):
+        files = []
+        pairs = (('net-eth0.cfg', DHCP_CONTENT_1),
+                 ('net-eth1.cfg', STATIC_CONTENT_1))
+
+        macs = {'eth1': 'b8:ae:ed:75:ff:2b',
+                'eth0': 'b8:ae:ed:75:ff:2a'}
+
+        dhcp = copy.deepcopy(DHCP_EXPECTED_1)
+        dhcp['mac_address'] = macs['eth0']
+
+        static = copy.deepcopy(STATIC_EXPECTED_1)
+        static['mac_address'] = macs['eth1']
+
+        expected = {'version': 1, 'config': [dhcp, static]}
+        with util.tempdir() as tmpd:
+            for fname, content in pairs:
+                fp = os.path.join(tmpd, fname)
+                files.append(fp)
+                util.write_file(fp, content)
+
+            found = net.config_from_klibc_net_cfg(files=files, mac_addrs=macs)
+            self.assertEqual(found, expected)
+
+    def test_cmdline_with_b64(self):
+        data = base64.b64encode(json.dumps(self.simple_cfg).encode())
+        encoded_text = data.decode()
+        cmdline = 'ro network-config=' + encoded_text + ' root=foo'
+        found = net.read_kernel_cmdline_config(cmdline=cmdline)
+        self.assertEqual(found, self.simple_cfg)
+
+    def test_cmdline_with_b64_gz(self):
+        data = _gzip_data(json.dumps(self.simple_cfg).encode())
+        encoded_text = base64.b64encode(data).decode()
+        cmdline = 'ro network-config=' + encoded_text + ' root=foo'
+        found = net.read_kernel_cmdline_config(cmdline=cmdline)
+        self.assertEqual(found, self.simple_cfg)
+
+
+def _gzip_data(data):
+    with io.BytesIO() as iobuf:
+        gzfp = gzip.GzipFile(mode="wb", fileobj=iobuf)
+        gzfp.write(data)
+        gzfp.close()
+        return iobuf.getvalue()