aif-ng/aif/network/networkd.py

import logging
import os
##
# We have to use Jinja2 because while there are ways to *parse* an INI with duplicate keys
# (https://stackoverflow.com/a/38286559/733214), there's no way to *write* an INI with them using configparser.
# So we use Jinja2 logic.
import jinja2
##
import aif.utils
from . import _common


_logger = logging.getLogger(__name__)


class Connection(_common.BaseConnection):
    def __init__(self, iface_xml):
        super().__init__(iface_xml)
        self.provider_type = 'systemd-networkd'
        self.packages = set()
        self.services = {
            ('/usr/lib/systemd/system/systemd-networkd.service'): ('etc/systemd/system/'
                                                                   'multi-user.target.wants/'
                                                                   'systemd-networkd.service'),
            ('/usr/lib/systemd/system/systemd-networkd.service'): ('etc/systemd/system/'
                                                                   'dbus-org.freedesktop.network1.service'),
            ('/usr/lib/systemd/system/systemd-networkd.socket'): ('etc/systemd/system/'
                                                                  'sockets.target.wants/systemd-networkd.socket'),
            ('/usr/lib/systemd/system/systemd-networkd.socket'): ('etc/systemd/system/'
                                                                  'network-online.target.wants/'
                                                                  'systemd-networkd-wait-online.service'),
            # We include these *even if* self.auto['resolvers'][*] are false.
            ('/usr/lib/systemd/system/systemd-resolved.service'): ('etc/systemd/system/'
                                                                   'dbus-org.freedesktop.resolve1.service'),
            ('/usr/lib/systemd/system/systemd-resolved.service'): ('etc/systemd/'
                                                                   'system/multi-user.target.wants/'
                                                                   'systemd-resolved.service')}
        self._wpasupp = {}
        self._initJ2()

    def _initCfg(self):
        _logger.info('Building config.')
        if self.device == 'auto':
            self.device = _common.getDefIface(self.connection_type)
        self._cfg = {'Match': {'Name': self.device},
                     'Network': {'Description': ('A {0} profile for {1} '
                                                 '(generated by AIF-NG)').format(self.connection_type,
                                                                                 self.device),
                                 'DefaultRouteOnDevice': ('true' if self.is_defroute else 'false'),
                                 # This (may) get modified by logic below.
                                 'IPv6AcceptRA': 'false',
                                 'LinkLocalAddressing': 'no'}}
        if self.domain:
            self._cfg['Network']['Domains'] = self.domain
        if self.resolvers:
            self._cfg['Network']['DNS'] = [str(ip) for ip in self.resolvers]
        if all((self.auto['addresses']['ipv4'], self.auto['addresses']['ipv6'])):
            self._cfg['Network']['IPv6AcceptRA'] = 'true'
            self._cfg['Network']['LinkLocalAddressing'] = 'ipv6'
            self._cfg['Network']['DHCP'] = 'yes'
        elif self.auto['addresses']['ipv4'] and not self.auto['addresses']['ipv6']:
            self._cfg['Network']['DHCP'] = 'ipv4'
        elif (not self.auto['addresses']['ipv4']) and self.auto['addresses']['ipv6']:
            self._cfg['Network']['IPv6AcceptRA'] = 'true'
            self._cfg['Network']['LinkLocalAddressing'] = 'ipv6'
            self._cfg['Network']['DHCP'] = 'ipv6'
        else:
            self._cfg['Network']['DHCP'] = 'no'
        if any((self.auto['addresses']['ipv4'], self.auto['routes']['ipv4'], self.auto['resolvers']['ipv4'])):
            t = 'ipv4'
            self._cfg['DHCPv4'] = {'UseDNS': ('true' if self.auto['resolvers'][t] else 'false'),
                                   'UseRoutes': ('true' if self.auto['routes'][t] else 'false')}
        if any((self.auto['addresses']['ipv6'], self.auto['routes']['ipv6'], self.auto['resolvers']['ipv6'])):
            t = 'ipv6'
            self._cfg['Network']['IPv6AcceptRA'] = 'true'
            self._cfg['DHCPv6'] = {'UseDNS': ('true' if self.auto['resolvers'][t] else 'false')}
        for t in ('ipv4', 'ipv6'):
            if self.addrs[t]:
                if t == 'ipv6':
                    self._cfg['Network']['LinkLocalAddressing'] = 'ipv6'
                if 'Address' not in self._cfg.keys():
                    self._cfg['Address'] = []
                for addr, net, gw in self.addrs[t]:
                    a = {'Address': '{0}/{1}'.format(str(addr), str(net.prefixlen))}
                    self._cfg['Address'].append(a)
            if self.routes[t]:
                if 'Route' not in self._cfg.keys():
                    self._cfg['Route'] = []
                for route, net, gw in self.routes[t]:
                    r = {'Gateway': str(gw),
                         'Destination': '{0}/{1}'.format(str(route), str(net.prefixlen))}
                    self._cfg['Route'].append(r)
        if self._cfg['Network']['IPv6AcceptRA'] == 'true':
            self._cfg['Network']['LinkLocalAddressing'] = 'ipv6'
            if 'IPv6AcceptRA' not in self._cfg.keys():
                self._cfg['IPv6AcceptRA'] = {'UseDNS': ('true' if self.auto['resolvers']['ipv6'] else 'false')}
        self._initConnCfg()
        _logger.info('Config built successfully.')
        return(None)

    def _initJ2(self):
        _logger.debug('Fetching template from networkd.conf.j2')
        self.j2_env = jinja2.Environment(loader = jinja2.FileSystemLoader(searchpath = './'))
        self.j2_env.filters.update(aif.utils.j2_filters)
        self.j2_tpl = self.j2_env.get_template('networkd.conf.j2')
        return(None)

    def writeConf(self, chroot_base):
        cfgroot = os.path.join(chroot_base, 'etc', 'systemd', 'network')
        cfgfile = os.path.join(cfgroot, self.id)
        os.makedirs(cfgroot, exist_ok = True)
        os.chown(cfgroot, 0, 0)
        os.chmod(cfgroot, 0o0755)
        with open(cfgfile, 'w') as fh:
            fh.write(self.j2_tpl.render(cfg = self._cfg))
        os.chmod(cfgfile, 0o0644)
        os.chown(cfgfile, 0, 0)
        self._writeConnCfg(chroot_base)
        _logger.info('Wrote: {0}'.format(cfgfile))
        _logger.debug('Rendering variables: {0}'.format(self._cfg))
        _logger.debug('Rendered template: {0}'.format(self.j2_tpl.render(cfg = self._cfg)))
        return(None)


class Ethernet(Connection):
    def __init__(self, iface_xml):
        super().__init__(iface_xml)
        self.connection_type = 'ethernet'
        self._initCfg()


class Wireless(Connection):
    def __init__(self, iface_xml):
        super().__init__(iface_xml)
        self.connection_type = 'wireless'
        self.packages.add('wpa_supplicant')
        self.services['src'] = 'dest'
        self._initCfg()

    def _initConnCfg(self):
        self._wpasupp['ssid'] = '"{0}"'.format(self.xml.attrib['essid'])
        hidden = aif.utils.xmlBool(self.xml.attrib.get('hidden', 'false'))
        if hidden:
            self._wpasupp['scan_ssid'] = 1
        try:
            bssid = self.xml.attrib.get('bssid').strip()
        except AttributeError:
            bssid = None
        if bssid:
            bssid = _common.canonizeEUI(bssid)
            self._wpasupp['bssid'] = bssid
            self._wpasupp['bssid_whitelist'] = bssid
        crypto = self.xml.find('encryption')
        if crypto:
            crypto = _common.convertWifiCrypto(crypto, self._cfg['BASE']['ESSID'])
            # if crypto['type'] in ('wpa', 'wpa2', 'wpa3'):
            # TODO: WPA2 enterprise
            if crypto['type'] in ('wpa', 'wpa2'):
                self._wpasupp['psk'] = crypto['auth']['psk']
        else:
            self._wpasupp['key_mgmt'] = 'NONE'
        _logger.debug('Fetching template from wpa_supplicant.conf.j2')
        self.wpasupp_tpl = self.j2_env.get_template('wpa_supplicant.conf.j2')
        self.services[('/usr/lib/systemd/system/wpa_supplicant@.service')] = ('etc/systemd/'
                                                                              'system/'
                                                                              'multi-user.target.wants/'
                                                                              'wpa_supplicant@'
                                                                              '{0}.service').format(self.device)
        return(None)

    def _writeConnCfg(self, chroot_base):
        cfgroot = os.path.join(chroot_base, 'etc', 'wpa_supplicant')
        cfgfile = os.path.join(cfgroot, 'wpa_supplicant-{0}.conf'.format(self.device))
        os.makedirs(cfgroot, exist_ok = True)
        os.chown(cfgroot, 0, 0)
        os.chmod(cfgroot, 0o0755)
        with open(cfgfile, 'w') as fh:
            fh.write(self.wpasupp_tpl.render(wpa = self._wpasupp))
        os.chown(cfgfile, 0, 0)
        os.chmod(cfgfile, 0o0640)
        _logger.info('Wrote: {0}'.format(cfgfile))
        _logger.debug('Rendering variables: {0}'.format(self._wpasupp))
        _logger.debug('Rendered template: {0}'.format(self.wpasupp_tpl.render(wpa = self._wpasupp)))
        return(None)