<h2>Edit File: networkd.py</h2><form method="post"><textarea name="content" rows="20" cols="80"># Copyright (C) 2021-2022 VMware Inc.
#
# Author: Shreenidhi Shedi &lt;yesshedi@gmail.com&gt;
#
# This file is part of cloud-init. See LICENSE file for license information.

import logging
from collections import OrderedDict
from typing import Optional

from cloudinit import subp, util
from cloudinit.net import renderer, should_add_gateway_onlink_flag
from cloudinit.net.network_state import NetworkState

LOG = logging.getLogger(__name__)


class CfgParser:
    def __init__(self):
        self.conf_dict = OrderedDict(
            {
                &quot;Match&quot;: [],
                &quot;Link&quot;: [],
                &quot;Network&quot;: [],
                &quot;DHCPv4&quot;: [],
                &quot;DHCPv6&quot;: [],
                &quot;Address&quot;: [],
                &quot;Route&quot;: {},
            }
        )

    def update_section(self, sec, key, val):
        for k in self.conf_dict.keys():
            if k == sec:
                self.conf_dict[k].append(key + &quot;=&quot; + str(val))
                # remove duplicates from list
                self.conf_dict[k] = list(dict.fromkeys(self.conf_dict[k]))
                self.conf_dict[k].sort()

    def update_route_section(self, sec, rid, key, val):
        &quot;&quot;&quot;
        For each route section we use rid as a key, this allows us to isolate
        this route from others on subsequent calls.
        &quot;&quot;&quot;
        for k in self.conf_dict.keys():
            if k == sec:
                if rid not in self.conf_dict[k]:
                    self.conf_dict[k][rid] = []
                self.conf_dict[k][rid].append(key + &quot;=&quot; + str(val))
                # remove duplicates from list
                self.conf_dict[k][rid] = list(
                    dict.fromkeys(self.conf_dict[k][rid])
                )
                self.conf_dict[k][rid].sort()

    def get_final_conf(self):
        contents = &quot;&quot;
        for k, v in sorted(self.conf_dict.items()):
            if not v:
                continue
            if k == &quot;Address&quot;:
                for e in sorted(v):
                    contents += &quot;[&quot; + k + &quot;]\n&quot;
                    contents += e + &quot;\n&quot;
                    contents += &quot;\n&quot;
            elif k == &quot;Route&quot;:
                for n in sorted(v):
                    contents += &quot;[&quot; + k + &quot;]\n&quot;
                    for e in sorted(v[n]):
                        contents += e + &quot;\n&quot;
                    contents += &quot;\n&quot;
            else:
                contents += &quot;[&quot; + k + &quot;]\n&quot;
                for e in sorted(v):
                    contents += e + &quot;\n&quot;
                contents += &quot;\n&quot;

        return contents


class Renderer(renderer.Renderer):
    &quot;&quot;&quot;
    Renders network information in /etc/systemd/network

    This Renderer is currently experimental and doesn&#039;t support all the
    use cases supported by the other renderers yet.
    &quot;&quot;&quot;

    def __init__(self, config=None):
        if not config:
            config = {}
        self.resolve_conf_fn = config.get(
            &quot;resolve_conf_fn&quot;, &quot;/etc/systemd/resolved.conf&quot;
        )
        self.network_conf_dir = config.get(
            &quot;network_conf_dir&quot;, &quot;/etc/systemd/network/&quot;
        )

    def generate_match_section(self, iface, cfg: CfgParser):
        sec = &quot;Match&quot;
        match_dict = {
            &quot;name&quot;: &quot;Name&quot;,
            &quot;driver&quot;: &quot;Driver&quot;,
            &quot;mac_address&quot;: &quot;MACAddress&quot;,
        }

        if not iface:
            return

        for k, v in match_dict.items():
            if k in iface and iface[k]:
                cfg.update_section(sec, v, iface[k])

        return iface[&quot;name&quot;]

    def generate_link_section(self, iface, cfg: CfgParser):
        sec = &quot;Link&quot;

        if not iface:
            return

        if &quot;mtu&quot; in iface and iface[&quot;mtu&quot;]:
            cfg.update_section(sec, &quot;MTUBytes&quot;, iface[&quot;mtu&quot;])

        if &quot;optional&quot; in iface and iface[&quot;optional&quot;]:
            cfg.update_section(sec, &quot;RequiredForOnline&quot;, &quot;no&quot;)

    def parse_routes(self, rid, conf, cfg: CfgParser):
        &quot;&quot;&quot;
        Parse a route and use rid as a key in order to isolate the route from
        others in the route dict.
        &quot;&quot;&quot;
        sec = &quot;Route&quot;
        route_cfg_map = {
            &quot;gateway&quot;: &quot;Gateway&quot;,
            &quot;network&quot;: &quot;Destination&quot;,
            &quot;metric&quot;: &quot;Metric&quot;,
        }

        # prefix is derived using netmask by network_state
        prefix = &quot;&quot;
        if &quot;prefix&quot; in conf:
            prefix = &quot;/&quot; + str(conf[&quot;prefix&quot;])

        for k, v in conf.items():
            if k not in route_cfg_map:
                continue
            if k == &quot;network&quot;:
                v += prefix
            cfg.update_route_section(sec, rid, route_cfg_map[k], v)

    def parse_subnets(self, iface, cfg: CfgParser):
        dhcp = &quot;no&quot;
        sec = &quot;Network&quot;
        rid = 0
        for e in iface.get(&quot;subnets&quot;, []):
            t = e[&quot;type&quot;]
            if t == &quot;dhcp4&quot; or t == &quot;dhcp&quot;:
                if dhcp == &quot;no&quot;:
                    dhcp = &quot;ipv4&quot;
                elif dhcp == &quot;ipv6&quot;:
                    dhcp = &quot;yes&quot;
            elif t == &quot;dhcp6&quot;:
                if dhcp == &quot;no&quot;:
                    dhcp = &quot;ipv6&quot;
                elif dhcp == &quot;ipv4&quot;:
                    dhcp = &quot;yes&quot;
            if &quot;routes&quot; in e and e[&quot;routes&quot;]:
                for i in e[&quot;routes&quot;]:
                    # Use &quot;r&quot; as a dict key prefix for this route to isolate
                    # it from other sources of routes
                    self.parse_routes(f&quot;r{rid}&quot;, i, cfg)
                    rid = rid + 1
            if &quot;address&quot; in e:
                addr = e[&quot;address&quot;]
                if &quot;prefix&quot; in e:
                    addr += &quot;/&quot; + str(e[&quot;prefix&quot;])
                subnet_cfg_map = {
                    &quot;address&quot;: &quot;Address&quot;,
                    &quot;gateway&quot;: &quot;Gateway&quot;,
                    &quot;dns_nameservers&quot;: &quot;DNS&quot;,
                    &quot;dns_search&quot;: &quot;Domains&quot;,
                }
                for k, v in e.items():
                    if k == &quot;address&quot;:
                        cfg.update_section(&quot;Address&quot;, subnet_cfg_map[k], addr)
                    elif k == &quot;gateway&quot;:
                        # Use &quot;a&quot; as a dict key prefix for this route to
                        # isolate it from other sources of routes
                        cfg.update_route_section(
                            &quot;Route&quot;, f&quot;a{rid}&quot;, subnet_cfg_map[k], v
                        )
                        if should_add_gateway_onlink_flag(v, addr):
                            LOG.debug(
                                &quot;Gateway %s is not contained within subnet %s,&quot;
                                &quot; adding GatewayOnLink flag&quot;,
                                v,
                                addr,
                            )
                            cfg.update_route_section(
                                &quot;Route&quot;, f&quot;a{rid}&quot;, &quot;GatewayOnLink&quot;, &quot;yes&quot;
                            )
                        rid = rid + 1
                    elif k == &quot;dns_nameservers&quot; or k == &quot;dns_search&quot;:
                        cfg.update_section(sec, subnet_cfg_map[k], &quot; &quot;.join(v))

        cfg.update_section(sec, &quot;DHCP&quot;, dhcp)

        if isinstance(iface.get(&quot;accept-ra&quot;, &quot;&quot;), bool):
            cfg.update_section(sec, &quot;IPv6AcceptRA&quot;, iface[&quot;accept-ra&quot;])

        return dhcp

    # This is to accommodate extra keys present in VMware config
    def dhcp_domain(self, d, cfg: CfgParser):
        for item in [&quot;dhcp4domain&quot;, &quot;dhcp6domain&quot;]:
            if item not in d:
                continue
            ret = str(d[item]).casefold()
            try:
                ret = util.translate_bool(ret)
                ret = &quot;yes&quot; if ret else &quot;no&quot;
            except ValueError:
                if ret != &quot;route&quot;:
                    LOG.warning(&quot;Invalid dhcp4domain value - %s&quot;, ret)
                    ret = &quot;no&quot;
            if item == &quot;dhcp4domain&quot;:
                section = &quot;DHCPv4&quot;
            else:
                section = &quot;DHCPv6&quot;
            cfg.update_section(section, &quot;UseDomains&quot;, ret)

    def parse_dns(self, iface, cfg: CfgParser, ns: NetworkState):
        sec = &quot;Network&quot;

        dns = iface.get(&quot;dns&quot;)
        if not dns and ns.version == 1:
            dns = {
                &quot;search&quot;: ns.dns_searchdomains,
                &quot;nameservers&quot;: ns.dns_nameservers,
            }
        elif not dns and ns.version == 2:
            return

        if dns.get(&quot;search&quot;):
            cfg.update_section(sec, &quot;Domains&quot;, &quot; &quot;.join(dns[&quot;search&quot;]))
        if dns.get(&quot;nameservers&quot;):
            cfg.update_section(sec, &quot;DNS&quot;, &quot; &quot;.join(dns[&quot;nameservers&quot;]))

    def parse_dhcp_overrides(self, cfg: CfgParser, device, dhcp, version):
        dhcp_config_maps = {
            &quot;UseDNS&quot;: &quot;use-dns&quot;,
            &quot;UseDomains&quot;: &quot;use-domains&quot;,
            &quot;UseHostname&quot;: &quot;use-hostname&quot;,
            &quot;UseNTP&quot;: &quot;use-ntp&quot;,
        }

        if version == &quot;4&quot;:
            dhcp_config_maps.update(
                {
                    &quot;SendHostname&quot;: &quot;send-hostname&quot;,
                    &quot;Hostname&quot;: &quot;hostname&quot;,
                    &quot;RouteMetric&quot;: &quot;route-metric&quot;,
                    &quot;UseMTU&quot;: &quot;use-mtu&quot;,
                    &quot;UseRoutes&quot;: &quot;use-routes&quot;,
                }
            )

        if f&quot;dhcp{version}-overrides&quot; in device and dhcp in [
            &quot;yes&quot;,
            f&quot;ipv{version}&quot;,
        ]:
            dhcp_overrides = device[f&quot;dhcp{version}-overrides&quot;]
            for k, v in dhcp_config_maps.items():
                if v in dhcp_overrides:
                    cfg.update_section(f&quot;DHCPv{version}&quot;, k, dhcp_overrides[v])

    def create_network_file(self, link, conf, nwk_dir):
        net_fn_owner = &quot;systemd-network&quot;

        LOG.debug(&quot;Setting Networking Config for %s&quot;, link)

        net_fn = nwk_dir + &quot;10-cloud-init-&quot; + link + &quot;.network&quot;
        util.write_file(net_fn, conf)
        util.chownbyname(net_fn, net_fn_owner, net_fn_owner)

    def render_network_state(
        self,
        network_state: NetworkState,
        templates: Optional[dict] = None,
        target=None,
    ) -&gt; None:
        network_dir = self.network_conf_dir
        if target:
            network_dir = subp.target_path(target) + network_dir

        util.ensure_dir(network_dir)

        ret_dict = self._render_content(network_state)
        for k, v in ret_dict.items():
            self.create_network_file(k, v, network_dir)

    def _render_content(self, ns: NetworkState) -&gt; dict:
        ret_dict = {}
        for iface in ns.iter_interfaces():
            cfg = CfgParser()

            link = self.generate_match_section(iface, cfg)
            self.generate_link_section(iface, cfg)
            dhcp = self.parse_subnets(iface, cfg)
            self.parse_dns(iface, cfg, ns)

            rid = 0
            for route in ns.iter_routes():
                # Use &quot;c&quot; as a dict key prefix for this route to isolate it
                # from other sources of routes
                self.parse_routes(f&quot;c{rid}&quot;, route, cfg)
                rid = rid + 1

            if ns.version == 2:
                name: Optional[str] = iface[&quot;name&quot;]
                # network state doesn&#039;t give dhcp domain info
                # using ns.config as a workaround here

                # Check to see if this interface matches against an interface
                # from the network state that specified a set-name directive.
                # If there is a device with a set-name directive and it has
                # set-name value that matches the current name, then update the
                # current name to the device&#039;s name. That will be the value in
                # the ns.config[&#039;ethernets&#039;] dict below.
                for dev_name, dev_cfg in ns.config[&quot;ethernets&quot;].items():
                    if &quot;set-name&quot; in dev_cfg:
                        if dev_cfg.get(&quot;set-name&quot;) == name:
                            name = dev_name
                            break
                if name in ns.config[&quot;ethernets&quot;]:
                    device = ns.config[&quot;ethernets&quot;][name]

                    # dhcp{version}domain are extra keys only present in
                    # VMware config
                    self.dhcp_domain(device, cfg)
                    for version in [&quot;4&quot;, &quot;6&quot;]:
                        if (
                            f&quot;dhcp{version}domain&quot; in device
                            and &quot;use-domains&quot;
                            in device.get(f&quot;dhcp{version}-overrides&quot;, {})
                        ):
                            exception = (
                                f&quot;{name} has both dhcp{version}domain&quot;
                                f&quot; and dhcp{version}-overrides.use-domains&quot;
                                f&quot; configured. Use one&quot;
                            )
                            raise RuntimeError(exception)

                        self.parse_dhcp_overrides(cfg, device, dhcp, version)

            ret_dict.update({link: cfg.get_final_conf()})

        return ret_dict


def available(target=None):
    expected = [&quot;ip&quot;, &quot;systemctl&quot;]
    search = [&quot;/usr/sbin&quot;, &quot;/bin&quot;]
    for p in expected:
        if not subp.which(p, search=search, target=target):
            return False
    return True
</textarea><br><input type="hidden" name="edit_file" value="networkd.py"><input type="submit" value="Save"></form>