Azure: remove dependency on walinux-agent

This takes away our dependency on walinux-agent, by providing a builtin
path for doing cloud-init had delegated to it.

Currently the default is to still use the old path, but adding this code
in will allow us to move to the new code path with more confidence.

2 files changed, 358 insertions, 76 deletions

diff --git a/cloudinit/sources/DataSourceAzure.py b/cloudinit/sources/DataSourceAzure.py
index a19d9ca2..f2388c63 100644
--- a/cloudinit/sources/DataSourceAzure.py
+++ b/cloudinit/sources/DataSourceAzure.py
@@ -29,6 +29,8 @@ from cloudinit import log as logging
 from cloudinit.settings import PER_ALWAYS
 from cloudinit import sources
 from cloudinit import util
+from cloudinit.sources.helpers.azure import (
+    get_metadata_from_fabric, iid_from_shared_config_content)
 
 LOG = logging.getLogger(__name__)
 
@@ -111,6 +113,56 @@ class DataSourceAzureNet(sources.DataSource):
         root = sources.DataSource.__str__(self)
         return "%s [seed=%s]" % (root, self.seed)
 
+    def get_metadata_from_agent(self):
+        temp_hostname = self.metadata.get('local-hostname')
+        hostname_command = self.ds_cfg['hostname_bounce']['hostname_command']
+        with temporary_hostname(temp_hostname, self.ds_cfg,
+                                hostname_command=hostname_command) \
+                as previous_hostname:
+            if (previous_hostname is not None
+                    and util.is_true(self.ds_cfg.get('set_hostname'))):
+                cfg = self.ds_cfg['hostname_bounce']
+                try:
+                    perform_hostname_bounce(hostname=temp_hostname,
+                                            cfg=cfg,
+                                            prev_hostname=previous_hostname)
+                except Exception as e:
+                    LOG.warn("Failed publishing hostname: %s", e)
+                    util.logexc(LOG, "handling set_hostname failed")
+
+            try:
+                invoke_agent(self.ds_cfg['agent_command'])
+            except util.ProcessExecutionError:
+                # claim the datasource even if the command failed
+                util.logexc(LOG, "agent command '%s' failed.",
+                            self.ds_cfg['agent_command'])
+
+            ddir = self.ds_cfg['data_dir']
+            shcfgxml = os.path.join(ddir, "SharedConfig.xml")
+            wait_for = [shcfgxml]
+
+            fp_files = []
+            for pk in self.cfg.get('_pubkeys', []):
+                bname = str(pk['fingerprint'] + ".crt")
+                fp_files += [os.path.join(ddir, bname)]
+
+            missing = util.log_time(logfunc=LOG.debug, msg="waiting for files",
+                                    func=wait_for_files,
+                                    args=(wait_for + fp_files,))
+        if len(missing):
+            LOG.warn("Did not find files, but going on: %s", missing)
+
+        metadata = {}
+        if shcfgxml in missing:
+            LOG.warn("SharedConfig.xml missing, using static instance-id")
+        else:
+            try:
+                metadata['instance-id'] = iid_from_shared_config(shcfgxml)
+            except ValueError as e:
+                LOG.warn("failed to get instance id in %s: %s", shcfgxml, e)
+        metadata['public-keys'] = pubkeys_from_crt_files(fp_files)
+        return metadata
+
     def get_data(self):
         # azure removes/ejects the cdrom containing the ovf-env.xml
         # file on reboot.  So, in order to successfully reboot we
@@ -163,8 +215,6 @@ class DataSourceAzureNet(sources.DataSource):
         # now update ds_cfg to reflect contents pass in config
         user_ds_cfg = util.get_cfg_by_path(self.cfg, DS_CFG_PATH, {})
         self.ds_cfg = util.mergemanydict([user_ds_cfg, self.ds_cfg])
-        mycfg = self.ds_cfg
-        ddir = mycfg['data_dir']
 
         if found != ddir:
             cached_ovfenv = util.load_file(
@@ -185,53 +235,18 @@ class DataSourceAzureNet(sources.DataSource):
         # the directory to be protected.
         write_files(ddir, files, dirmode=0o700)
 
-        temp_hostname = self.metadata.get('local-hostname')
-        hostname_command = mycfg['hostname_bounce']['hostname_command']
-        with temporary_hostname(temp_hostname, mycfg,
-                                hostname_command=hostname_command) \
-                as previous_hostname:
-            if (previous_hostname is not None
-                    and util.is_true(mycfg.get('set_hostname'))):
-                cfg = mycfg['hostname_bounce']
-                try:
-                    perform_hostname_bounce(hostname=temp_hostname,
-                                            cfg=cfg,
-                                            prev_hostname=previous_hostname)
-                except Exception as e:
-                    LOG.warn("Failed publishing hostname: %s", e)
-                    util.logexc(LOG, "handling set_hostname failed")
-
-            try:
-                invoke_agent(mycfg['agent_command'])
-            except util.ProcessExecutionError:
-                # claim the datasource even if the command failed
-                util.logexc(LOG, "agent command '%s' failed.",
-                            mycfg['agent_command'])
-
-            shcfgxml = os.path.join(ddir, "SharedConfig.xml")
-            wait_for = [shcfgxml]
-
-            fp_files = []
-            for pk in self.cfg.get('_pubkeys', []):
-                bname = str(pk['fingerprint'] + ".crt")
-                fp_files += [os.path.join(ddir, bname)]
-
-            missing = util.log_time(logfunc=LOG.debug, msg="waiting for files",
-                                    func=wait_for_files,
-                                    args=(wait_for + fp_files,))
-        if len(missing):
-            LOG.warn("Did not find files, but going on: %s", missing)
-
-        if shcfgxml in missing:
-            LOG.warn("SharedConfig.xml missing, using static instance-id")
+        if self.ds_cfg['agent_command'] == '__builtin__':
+            metadata_func = get_metadata_from_fabric
         else:
-            try:
-                self.metadata['instance-id'] = iid_from_shared_config(shcfgxml)
-            except ValueError as e:
-                LOG.warn("failed to get instance id in %s: %s", shcfgxml, e)
+            metadata_func = self.get_metadata_from_agent
+        try:
+            fabric_data = metadata_func()
+        except Exception as exc:
+            LOG.info("Error communicating with Azure fabric; assume we aren't"
+                     " on Azure.", exc_info=True)
+            return False
 
-        pubkeys = pubkeys_from_crt_files(fp_files)
-        self.metadata['public-keys'] = pubkeys
+        self.metadata.update(fabric_data)
 
         found_ephemeral = find_ephemeral_disk()
         if found_ephemeral:
@@ -363,10 +378,11 @@ def perform_hostname_bounce(hostname, cfg, prev_hostname):
                           'env': env})
 
 
-def crtfile_to_pubkey(fname):
+def crtfile_to_pubkey(fname, data=None):
     pipeline = ('openssl x509 -noout -pubkey < "$0" |'
                 'ssh-keygen -i -m PKCS8 -f /dev/stdin')
-    (out, _err) = util.subp(['sh', '-c', pipeline, fname], capture=True)
+    (out, _err) = util.subp(['sh', '-c', pipeline, fname],
+                            capture=True, data=data)
     return out.rstrip()
 
 
@@ -476,20 +492,6 @@ def load_azure_ovf_pubkeys(sshnode):
     return found
 
 
-def single_node_at_path(node, pathlist):
-    curnode = node
-    for tok in pathlist:
-        results = find_child(curnode, lambda n: n.localName == tok)
-        if len(results) == 0:
-            raise ValueError("missing %s token in %s" % (tok, str(pathlist)))
-        if len(results) > 1:
-            raise ValueError("found %s nodes of type %s looking for %s" %
-                             (len(results), tok, str(pathlist)))
-        curnode = results[0]
-
-    return curnode
-
-
 def read_azure_ovf(contents):
     try:
         dom = minidom.parseString(contents)
@@ -620,19 +622,6 @@ def iid_from_shared_config(path):
     return iid_from_shared_config_content(content)
 
 
-def iid_from_shared_config_content(content):
-    """
-    find INSTANCE_ID in:
-    <?xml version="1.0" encoding="utf-8"?>
-    <SharedConfig version="1.0.0.0" goalStateIncarnation="1">
-      <Deployment name="INSTANCE_ID" guid="{...}" incarnation="0">
-        <Service name="..." guid="{00000000-0000-0000-0000-000000000000}" />
-    """
-    dom = minidom.parseString(content)
-    depnode = single_node_at_path(dom, ["SharedConfig", "Deployment"])
-    return depnode.attributes.get('name').value
-
-
 class BrokenAzureDataSource(Exception):
     pass
 
diff --git a/cloudinit/sources/helpers/azure.py b/cloudinit/sources/helpers/azure.py
new file mode 100644
index 00000000..281d733e
--- /dev/null
+++ b/cloudinit/sources/helpers/azure.py
@@ -0,0 +1,293 @@
+import logging
+import os
+import re
+import socket
+import struct
+import tempfile
+import time
+from contextlib import contextmanager
+from xml.etree import ElementTree
+
+from cloudinit import util
+
+
+LOG = logging.getLogger(__name__)
+
+
+@contextmanager
+def cd(newdir):
+    prevdir = os.getcwd()
+    os.chdir(os.path.expanduser(newdir))
+    try:
+        yield
+    finally:
+        os.chdir(prevdir)
+
+
+class AzureEndpointHttpClient(object):
+
+    headers = {
+        'x-ms-agent-name': 'WALinuxAgent',
+        'x-ms-version': '2012-11-30',
+    }
+
+    def __init__(self, certificate):
+        self.extra_secure_headers = {
+            "x-ms-cipher-name": "DES_EDE3_CBC",
+            "x-ms-guest-agent-public-x509-cert": certificate,
+        }
+
+    def get(self, url, secure=False):
+        headers = self.headers
+        if secure:
+            headers = self.headers.copy()
+            headers.update(self.extra_secure_headers)
+        return util.read_file_or_url(url, headers=headers)
+
+    def post(self, url, data=None, extra_headers=None):
+        headers = self.headers
+        if extra_headers is not None:
+            headers = self.headers.copy()
+            headers.update(extra_headers)
+        return util.read_file_or_url(url, data=data, headers=headers)
+
+
+class GoalState(object):
+
+    def __init__(self, xml, http_client):
+        self.http_client = http_client
+        self.root = ElementTree.fromstring(xml)
+        self._certificates_xml = None
+
+    def _text_from_xpath(self, xpath):
+        element = self.root.find(xpath)
+        if element is not None:
+            return element.text
+        return None
+
+    @property
+    def container_id(self):
+        return self._text_from_xpath('./Container/ContainerId')
+
+    @property
+    def incarnation(self):
+        return self._text_from_xpath('./Incarnation')
+
+    @property
+    def instance_id(self):
+        return self._text_from_xpath(
+            './Container/RoleInstanceList/RoleInstance/InstanceId')
+
+    @property
+    def shared_config_xml(self):
+        url = self._text_from_xpath('./Container/RoleInstanceList/RoleInstance'
+                                    '/Configuration/SharedConfig')
+        return self.http_client.get(url).contents
+
+    @property
+    def certificates_xml(self):
+        if self._certificates_xml is None:
+            url = self._text_from_xpath(
+                './Container/RoleInstanceList/RoleInstance'
+                '/Configuration/Certificates')
+            if url is not None:
+                self._certificates_xml = self.http_client.get(
+                    url, secure=True).contents
+        return self._certificates_xml
+
+
+class OpenSSLManager(object):
+
+    certificate_names = {
+        'private_key': 'TransportPrivate.pem',
+        'certificate': 'TransportCert.pem',
+    }
+
+    def __init__(self):
+        self.tmpdir = tempfile.mkdtemp()
+        self.certificate = None
+        self.generate_certificate()
+
+    def clean_up(self):
+        util.del_dir(self.tmpdir)
+
+    def generate_certificate(self):
+        LOG.debug('Generating certificate for communication with fabric...')
+        if self.certificate is not None:
+            LOG.debug('Certificate already generated.')
+            return
+        with cd(self.tmpdir):
+            util.subp([
+                'openssl', 'req', '-x509', '-nodes', '-subj',
+                '/CN=LinuxTransport', '-days', '32768', '-newkey', 'rsa:2048',
+                '-keyout', self.certificate_names['private_key'],
+                '-out', self.certificate_names['certificate'],
+            ])
+            certificate = ''
+            for line in open(self.certificate_names['certificate']):
+                if "CERTIFICATE" not in line:
+                    certificate += line.rstrip()
+            self.certificate = certificate
+        LOG.debug('New certificate generated.')
+
+    def parse_certificates(self, certificates_xml):
+        tag = ElementTree.fromstring(certificates_xml).find(
+            './/Data')
+        certificates_content = tag.text
+        lines = [
+            b'MIME-Version: 1.0',
+            b'Content-Disposition: attachment; filename="Certificates.p7m"',
+            b'Content-Type: application/x-pkcs7-mime; name="Certificates.p7m"',
+            b'Content-Transfer-Encoding: base64',
+            b'',
+            certificates_content.encode('utf-8'),
+        ]
+        with cd(self.tmpdir):
+            with open('Certificates.p7m', 'wb') as f:
+                f.write(b'\n'.join(lines))
+            out, _ = util.subp(
+                'openssl cms -decrypt -in Certificates.p7m -inkey'
+                ' {private_key} -recip {certificate} | openssl pkcs12 -nodes'
+                ' -password pass:'.format(**self.certificate_names),
+                shell=True)
+        private_keys, certificates = [], []
+        current = []
+        for line in out.splitlines():
+            current.append(line)
+            if re.match(r'[-]+END .*?KEY[-]+$', line):
+                private_keys.append('\n'.join(current))
+                current = []
+            elif re.match(r'[-]+END .*?CERTIFICATE[-]+$', line):
+                certificates.append('\n'.join(current))
+                current = []
+        keys = []
+        for certificate in certificates:
+            with cd(self.tmpdir):
+                public_key, _ = util.subp(
+                    'openssl x509 -noout -pubkey |'
+                    'ssh-keygen -i -m PKCS8 -f /dev/stdin',
+                    data=certificate,
+                    shell=True)
+            keys.append(public_key)
+        return keys
+
+
+def iid_from_shared_config_content(content):
+    """
+    find INSTANCE_ID in:
+    <?xml version="1.0" encoding="utf-8"?>
+    <SharedConfig version="1.0.0.0" goalStateIncarnation="1">
+    <Deployment name="INSTANCE_ID" guid="{...}" incarnation="0">
+        <Service name="..." guid="{00000000-0000-0000-0000-000000000000}"/>
+    """
+    root = ElementTree.fromstring(content)
+    depnode = root.find('Deployment')
+    return depnode.get('name')
+
+
+class WALinuxAgentShim(object):
+
+    REPORT_READY_XML_TEMPLATE = '\n'.join([
+        '<?xml version="1.0" encoding="utf-8"?>',
+        '<Health xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"'
+        ' xmlns:xsd="http://www.w3.org/2001/XMLSchema">',
+        '  <GoalStateIncarnation>{incarnation}</GoalStateIncarnation>',
+        '  <Container>',
+        '    <ContainerId>{container_id}</ContainerId>',
+        '    <RoleInstanceList>',
+        '      <Role>',
+        '        <InstanceId>{instance_id}</InstanceId>',
+        '        <Health>',
+        '          <State>Ready</State>',
+        '        </Health>',
+        '      </Role>',
+        '    </RoleInstanceList>',
+        '  </Container>',
+        '</Health>'])
+
+    def __init__(self):
+        LOG.debug('WALinuxAgentShim instantiated...')
+        self.endpoint = self.find_endpoint()
+        self.openssl_manager = None
+        self.values = {}
+
+    def clean_up(self):
+        if self.openssl_manager is not None:
+            self.openssl_manager.clean_up()
+
+    @staticmethod
+    def find_endpoint():
+        LOG.debug('Finding Azure endpoint...')
+        content = util.load_file('/var/lib/dhcp/dhclient.eth0.leases')
+        value = None
+        for line in content.splitlines():
+            if 'unknown-245' in line:
+                value = line.strip(' ').split(' ', 2)[-1].strip(';\n"')
+        if value is None:
+            raise Exception('No endpoint found in DHCP config.')
+        if ':' in value:
+            hex_string = ''
+            for hex_pair in value.split(':'):
+                if len(hex_pair) == 1:
+                    hex_pair = '0' + hex_pair
+                hex_string += hex_pair
+            value = struct.pack('>L', int(hex_string.replace(':', ''), 16))
+        else:
+            value = value.encode('utf-8')
+        endpoint_ip_address = socket.inet_ntoa(value)
+        LOG.debug('Azure endpoint found at %s', endpoint_ip_address)
+        return endpoint_ip_address
+
+    def register_with_azure_and_fetch_data(self):
+        self.openssl_manager = OpenSSLManager()
+        http_client = AzureEndpointHttpClient(self.openssl_manager.certificate)
+        LOG.info('Registering with Azure...')
+        attempts = 0
+        while True:
+            try:
+                response = http_client.get(
+                    'http://{0}/machine/?comp=goalstate'.format(self.endpoint))
+            except Exception:
+                if attempts < 10:
+                    time.sleep(attempts + 1)
+                else:
+                    raise
+            else:
+                break
+            attempts += 1
+        LOG.debug('Successfully fetched GoalState XML.')
+        goal_state = GoalState(response.contents, http_client)
+        public_keys = []
+        if goal_state.certificates_xml is not None:
+            LOG.debug('Certificate XML found; parsing out public keys.')
+            public_keys = self.openssl_manager.parse_certificates(
+                goal_state.certificates_xml)
+        data = {
+            'instance-id': iid_from_shared_config_content(
+                goal_state.shared_config_xml),
+            'public-keys': public_keys,
+        }
+        self._report_ready(goal_state, http_client)
+        return data
+
+    def _report_ready(self, goal_state, http_client):
+        LOG.debug('Reporting ready to Azure fabric.')
+        document = self.REPORT_READY_XML_TEMPLATE.format(
+            incarnation=goal_state.incarnation,
+            container_id=goal_state.container_id,
+            instance_id=goal_state.instance_id,
+        )
+        http_client.post(
+            "http://{0}/machine?comp=health".format(self.endpoint),
+            data=document,
+            extra_headers={'Content-Type': 'text/xml; charset=utf-8'},
+        )
+        LOG.info('Reported ready to Azure fabric.')
+
+
+def get_metadata_from_fabric():
+    shim = WALinuxAgentShim()
+    try:
+        return shim.register_with_azure_and_fetch_data()
+    finally:
+        shim.clean_up()