FM / sources

/snap/core18/2979/usr/lib/python3/dist-packages/cloudinit/sources/

/snap/core18/2979/usr/lib/python3/dist-packages/cloudinit/sources

Name	Size	Mode	Actions
azure/	-	0755	rm
helpers/	-	0755	rm
__pycache__/	-	0755	rm
DataSourceAliYun.py	2221	0644	edit dl rm
DataSourceAltCloud.py	8489	0644	edit dl rm
DataSourceAzure.py	70329	0644	edit dl rm
DataSourceBigstep.py	1964	0644	edit dl rm
DataSourceCloudSigma.py	4096	0644	edit dl rm
DataSourceCloudStack.py	10090	0644	edit dl rm
DataSourceConfigDrive.py	11275	0644	edit dl rm
DataSourceDigitalOcean.py	3904	0644	edit dl rm
DataSourceEc2.py	36670	0644	edit dl rm
DataSourceExoscale.py	9105	0644	edit dl rm
DataSourceGCE.py	12251	0644	edit dl rm
DataSourceHetzner.py	5537	0644	edit dl rm
DataSourceIBMCloud.py	14391	0644	edit dl rm
DataSourceLXD.py	17599	0644	edit dl rm
DataSourceMAAS.py	14735	0644	edit dl rm
DataSourceNoCloud.py	12676	0644	edit dl rm
DataSourceNone.py	1393	0644	edit dl rm
DataSourceNWCS.py	4768	0644	edit dl rm
DataSourceOpenNebula.py	16362	0644	edit dl rm
DataSourceOpenStack.py	10713	0644	edit dl rm
DataSourceOracle.py	15069	0644	edit dl rm
DataSourceOVF.py	10724	0644	edit dl rm
DataSourceRbxCloud.py	8021	0644	edit dl rm
DataSourceScaleway.py	9693	0644	edit dl rm
DataSourceSmartOS.py	34909	0644	edit dl rm
DataSourceUpCloud.py	5686	0644	edit dl rm
DataSourceVMware.py	33485	0644	edit dl rm
DataSourceVultr.py	4279	0644	edit dl rm
__init__.py	41771	0644	edit dl rm

Edit: /snap/core18/2979/usr/lib/python3/dist-packages/cloudinit/sources/DataSourceOVF.py (10724B)

# Copyright (C) 2011 Canonical Ltd.
# Copyright (C) 2012 Hewlett-Packard Development Company, L.P.
# Copyright (C) 2012 Yahoo! Inc.
#
# Author: Scott Moser <scott.moser@canonical.com>
# Author: Juerg Hafliger <juerg.haefliger@hp.com>
# Author: Joshua Harlow <harlowja@yahoo-inc.com>
#
# This file is part of cloud-init. See LICENSE file for license information.

import base64
import os
import re
from xml.dom import minidom

from cloudinit import log as logging
from cloudinit import safeyaml, sources, subp, util

LOG = logging.getLogger(__name__)

class DataSourceOVF(sources.DataSource):

dsname = "OVF"

def __init__(self, sys_cfg, distro, paths):
        sources.DataSource.__init__(self, sys_cfg, distro, paths)
        self.seed = None
        self.seed_dir = os.path.join(paths.seed_dir, "ovf")
        self.environment = None
        self.cfg = {}
        self.supported_seed_starts = ("/", "file://")
        self._network_config = None

def __str__(self):
        root = sources.DataSource.__str__(self)
        return "%s [seed=%s]" % (root, self.seed)

def _get_data(self):
        found = []
        md = {}
        ud = ""
        vd = ""

defaults = {
            "instance-id": "iid-dsovf",
        }

(seedfile, contents) = get_ovf_env(self.paths.seed_dir)

if seedfile:
            # Found a seed dir
            seed = os.path.join(self.paths.seed_dir, seedfile)
            (md, ud, cfg) = read_ovf_environment(contents)
            self.environment = contents
            found.append(seed)
        else:
            np = [
                ("com.vmware.guestInfo", transport_vmware_guestinfo),
                ("iso", transport_iso9660),
            ]
            name = None
            for name, transfunc in np:
                contents = transfunc()
                if contents:
                    break
            if contents:
                (md, ud, cfg) = read_ovf_environment(contents, True)
                self.environment = contents
                if "network-config" in md and md["network-config"]:
                    self._network_config = md["network-config"]
                found.append(name)

# There was no OVF transports found
        if len(found) == 0:
            return False

if "seedfrom" in md and md["seedfrom"]:
            seedfrom = md["seedfrom"]
            seedfound = False
            for proto in self.supported_seed_starts:
                if seedfrom.startswith(proto):
                    seedfound = proto
                    break
            if not seedfound:
                LOG.debug("Seed from %s not supported by %s", seedfrom, self)
                return False

(md_seed, ud, vd) = util.read_seeded(seedfrom, timeout=None)
            LOG.debug("Using seeded cache data from %s", seedfrom)

md = util.mergemanydict([md, md_seed])
            found.append(seedfrom)

# Now that we have exhausted any other places merge in the defaults
        md = util.mergemanydict([md, defaults])

self.seed = ",".join(found)
        self.metadata = md
        self.userdata_raw = ud
        self.vendordata_raw = vd
        self.cfg = cfg
        return True

def _get_subplatform(self):
        return "ovf (%s)" % self.seed

def get_public_ssh_keys(self):
        if "public-keys" not in self.metadata:
            return []
        pks = self.metadata["public-keys"]
        if isinstance(pks, (list)):
            return pks
        else:
            return [pks]

# The data sources' config_obj is a cloud-config formatted
    # object that came to it from ways other than cloud-config
    # because cloud-config content would be handled elsewhere
    def get_config_obj(self):
        return self.cfg

@property
    def network_config(self):
        return self._network_config

class DataSourceOVFNet(DataSourceOVF):
    def __init__(self, sys_cfg, distro, paths):
        DataSourceOVF.__init__(self, sys_cfg, distro, paths)
        self.seed_dir = os.path.join(paths.seed_dir, "ovf-net")
        self.supported_seed_starts = ("http://", "https://")

# This will return a dict with some content
#  meta-data, user-data, some config
def read_ovf_environment(contents, read_network=False):
    props = get_properties(contents)
    md = {}
    cfg = {}
    ud = None
    cfg_props = ["password"]
    md_props = ["seedfrom", "local-hostname", "public-keys", "instance-id"]
    network_props = ["network-config"]
    for (prop, val) in props.items():
        if prop == "hostname":
            prop = "local-hostname"
        if prop in md_props:
            md[prop] = val
        elif prop in cfg_props:
            cfg[prop] = val
        elif prop in network_props and read_network:
            try:
                network_config = base64.b64decode(val.encode())
                md[prop] = safeload_yaml_or_dict(network_config).get("network")
            except Exception:
                LOG.debug("Ignore network-config in wrong format")
        elif prop == "user-data":
            try:
                ud = base64.b64decode(val.encode())
            except Exception:
                ud = val.encode()
    return (md, ud, cfg)

# Returns tuple of filename (in 'dirname', and the contents of the file)
# on "not found", returns 'None' for filename and False for contents
def get_ovf_env(dirname):
    env_names = ("ovf-env.xml", "ovf_env.xml", "OVF_ENV.XML", "OVF-ENV.XML")
    for fname in env_names:
        full_fn = os.path.join(dirname, fname)
        if os.path.isfile(full_fn):
            try:
                contents = util.load_file(full_fn)
                return (fname, contents)
            except Exception:
                util.logexc(LOG, "Failed loading ovf file %s", full_fn)
    return (None, False)

def maybe_cdrom_device(devname):
    """Test if devname matches known list of devices which may contain iso9660
       filesystems.

Be helpful in accepting either knames (with no leading /dev/) or full path
    names, but do not allow paths outside of /dev/, like /dev/foo/bar/xxx.
    """
    if not devname:
        return False
    elif not isinstance(devname, str):
        raise ValueError("Unexpected input for devname: %s" % devname)

# resolve '..' and multi '/' elements
    devname = os.path.normpath(devname)

# drop leading '/dev/'
    if devname.startswith("/dev/"):
        # partition returns tuple (before, partition, after)
        devname = devname.partition("/dev/")[-1]

# ignore leading slash (/sr0), else fail on / in name (foo/bar/xvdc)
    if devname.startswith("/"):
        devname = devname.split("/")[-1]
    elif devname.count("/") > 0:
        return False

# if empty string
    if not devname:
        return False

# default_regex matches values in /lib/udev/rules.d/60-cdrom_id.rules
    # KERNEL!="sr[0-9]*|hd[a-z]|xvd*", GOTO="cdrom_end"
    default_regex = r"^(sr[0-9]+|hd[a-z]|xvd.*)"
    devname_regex = os.environ.get("CLOUD_INIT_CDROM_DEV_REGEX", default_regex)
    cdmatch = re.compile(devname_regex)

return cdmatch.match(devname) is not None

# Transport functions are called with no arguments and return
# either None (indicating not present) or string content of an ovf-env.xml
def transport_iso9660(require_iso=True):

# Go through mounts to see if it was already mounted
    mounts = util.mounts()
    for (dev, info) in mounts.items():
        fstype = info["fstype"]
        if fstype != "iso9660" and require_iso:
            continue
        if not maybe_cdrom_device(dev):
            continue
        mp = info["mountpoint"]
        (_fname, contents) = get_ovf_env(mp)
        if contents is not False:
            return contents

if require_iso:
        mtype = "iso9660"
    else:
        mtype = None

# generate a list of devices with mtype filesystem, filter by regex
    devs = [
        dev
        for dev in util.find_devs_with("TYPE=%s" % mtype if mtype else None)
        if maybe_cdrom_device(dev)
    ]
    for dev in devs:
        try:
            (_fname, contents) = util.mount_cb(dev, get_ovf_env, mtype=mtype)
        except util.MountFailedError:
            LOG.debug("%s not mountable as iso9660", dev)
            continue

if contents is not False:
            return contents

return None

def transport_vmware_guestinfo():
    rpctool = "vmware-rpctool"
    not_found = None
    if not subp.which(rpctool):
        return not_found
    cmd = [rpctool, "info-get guestinfo.ovfEnv"]
    try:
        out, _err = subp.subp(cmd)
        if out:
            return out
        LOG.debug("cmd %s exited 0 with empty stdout: %s", cmd, out)
    except subp.ProcessExecutionError as e:
        if e.exit_code != 1:
            LOG.warning("%s exited with code %d", rpctool, e.exit_code)
            LOG.debug(e)
    return not_found

def find_child(node, filter_func):
    ret = []
    if not node.hasChildNodes():
        return ret
    for child in node.childNodes:
        if filter_func(child):
            ret.append(child)
    return ret

def get_properties(contents):

dom = minidom.parseString(contents)
    if dom.documentElement.localName != "Environment":
        raise XmlError("No Environment Node")

if not dom.documentElement.hasChildNodes():
        raise XmlError("No Child Nodes")

envNsURI = "http://schemas.dmtf.org/ovf/environment/1"

# could also check here that elem.namespaceURI ==
    #   "http://schemas.dmtf.org/ovf/environment/1"
    propSections = find_child(
        dom.documentElement, lambda n: n.localName == "PropertySection"
    )

if len(propSections) == 0:
        raise XmlError("No 'PropertySection's")

props = {}
    propElems = find_child(
        propSections[0], (lambda n: n.localName == "Property")
    )

for elem in propElems:
        key = elem.attributes.getNamedItemNS(envNsURI, "key").value
        val = elem.attributes.getNamedItemNS(envNsURI, "value").value
        props[key] = val

return props

class XmlError(Exception):
    pass

# Used to match classes to dependencies
datasources = (
    (DataSourceOVF, (sources.DEP_FILESYSTEM,)),
    (DataSourceOVFNet, (sources.DEP_FILESYSTEM, sources.DEP_NETWORK)),
)

# Return a list of data sources that match this set of dependencies
def get_datasource_list(depends):
    return sources.list_from_depends(depends, datasources)

def safeload_yaml_or_dict(data):
    """
    The meta data could be JSON or YAML. Since YAML is a strict superset of
    JSON, we will unmarshal the data as YAML. If data is None then a new
    dictionary is returned.
    """
    if not data:
        return {}
    return safeyaml.load(data)

# vi: ts=4 expandtab

cmd: