interfaces2netplan

#!/usr/bin/env python
# interfaces2netplan (part of ossobv/vcutil) // wdoekes/2018-2021
#   // Public Domain
#
# Quick attempt to convert the simplest interfaces files (from ifupdown)
# to netplan.io YAML syntax. If it has the slightest hint that it might miss
# something, it will abort with some kind of error.
#
# 1. Enumerates all files in /etc/network/if-*.d and warns.
# 2. Concatenates all files in /etc/network/interfaces{,.d/*} and attempts
#    to convert the contents to a valid netplan.io YAML file.
#
# Example output:
#
#   network:
#     version: 2
#     renderer: networkd
#     ethernets:
#       enp3s0:
#        addresses:
#          - 10.100.1.38/24
#          - 10.100.1.39/24
#        gateway4: 10.100.1.1
#
# When content, write output to the yaml, like this:
#
#   interfaces2netplan > /etc/netplan/01-netcfg.yaml
#
# After a succesful 'netplan try' you should be able to reboot and then
# remove the 'ifupdown' package. Make sure you remove/rename the
# interfaces file firs though. Otherwise the ifupdown removal will
# ifdown your interfaces >:)
#
# See also: "netplan ifupdown-migrate" (which apparently exists)
#
from collections import OrderedDict
import fnmatch
import os
import re
import sys


class Line(object):
    def __init__(self, line, lineno, filename):
        self.line = line
        self.lineno = lineno
        self.filename = filename

    @property
    def location(self):
        return '{}:{}'.format(self.filename, self.lineno)


def _netmaskbits(netmask):
    mask = [int(i) for i in netmask.split('.')]
    if len(mask) == 1:
        return mask[0]

    mask = (mask[0] << 24 | mask[1] << 16 | mask[2] << 8 | mask[3])
    bits = 32
    while mask:
        if ((mask >> 1) << 1) == mask:
            bits -= 1
            mask >>= 1
        else:
            break
    mask <<= (32 - bits)
    assert (0xffffffff << (32 - bits) & 0xffffffff) == mask
    return bits


def _stderr(data):
    "Shortcut for print to stderr"
    if os.isatty(sys.stderr.fileno()):
        data = '\x1b[31;1m{}\x1b[0m'.format(data)
    sys.stderr.write(data)
    sys.stderr.flush()


def _stdout(data):
    "Shortcut for print to stdout"
    sys.stdout.write(data)
    sys.stdout.flush()


def list_all_files(path):
    "Return all file-type files in supplied path"
    return sorted([
        os.path.join(dir_, file_)
        for dir_, dirs, files in os.walk(path)
        for file_ in files])


class NetworkFiles(object):
    "Load up files in /etc/network and make sense of them"

    def __init__(self):
        self._interfaces = []
        self._ifupdownd = []
        self._unknown = []
        self._yaml = None
        self._populate(list_all_files('/etc/network'))

    def _populate(self, files):
        # Check that there is an interfaces file, and check whether we
        # expect files in interfaces.d. Also, make sure it is added first.
        self._interfaces.append('/etc/network/interfaces')
        with open(self._interfaces[0], 'r') as fp:
            lines = [i.strip() for i in fp.read().split('\n')]
            lines = [i for i in lines if i.startswith((
                'source ', 'source\t'))]
            if len(lines) == 0:
                # No 'source'? We see examples where the files are
                # configured/used anyway..
                interfaces_re = re.compile(
                    r'^/etc/network/interfaces.d/[^./][^/]*$')
            elif len(lines) == 1:
                # Check the source, and convert to regex.
                interfaces_re = re.compile(
                    fnmatch.translate(lines[0].split(None, 1)[-1]))
            else:
                raise ValueError('Multiple source lines found? {!r}'.format(
                    lines))

        # Check all supplied files against our lists/regexes.
        ifupdown_re = re.compile(
            r'^/etc/network/if-(down|post-down|pre-up|up)[.]d/[^/]*$')
        devnull = []
        destinations = (
            (self._ifupdownd, (lambda x: ifupdown_re.match(x))),
            (devnull, (lambda x: x == '/etc/network/interfaces')),
            (self._interfaces, (lambda x: interfaces_re.match(x))),
        )
        for file_ in files:
            for destlist, matches in destinations:
                if matches(file_):
                    destlist.append(file_)
                    break
            else:
                self._unknown.append(file_)

    def _get_appended_interfaces(self):
        lines = []

        for file_ in self._interfaces:
            with open(file_, 'r') as fp:
                data = fp.read()

                # Remove the one source entry we expect. We're doing
                # the sourcing here.
                if file_ == '/etc/network/interfaces':
                    data = '\n'.join([
                        ('' if line.strip().startswith((
                            'source ', 'source\t')) else line)
                        for line in data.split('\n')])

                lines.extend([
                    Line(line, (idx + 1), file_)
                    for idx, line in enumerate(data.split('\n'))])

        return lines

    def get_yaml(self):
        if self._yaml is None:
            interfaces_lines = self._get_appended_interfaces()
            ifile = InterfacesParser(interfaces_lines)
            self._yaml = ifile.to_netplan()
        return self._yaml

    def exitcode(self):
        # Don't even continue of we cannot generate a nice YAML.
        yaml = self.get_yaml()
        assert yaml is not None

        status = 0
        if self._ifupdownd:
            # Don't care about these "normal" if-up.d/if-down.d files.
            # They're generally not crucial.
            pass
        if self._unknown:
            # This can be worrisome, don't return 0/OK anymore.
            status = 2

        return status

    def show(self):
        yaml = self.get_yaml()
        _stdout(yaml)

    def hint(self):
        if os.isatty(sys.stdout.fileno()):
            # If you're redirecting stdout, you don't need this hint anymore.
            _stderr(
                "Example usage:\n"
                "  interfaces2netplan > /etc/netplan/01-netcfg.yaml\n\n")

    def warn(self):
        if self._ifupdownd:
            _stderr(
                "Found some files we do not convert, check manually:\n"
                "  {}\n\n".format('\n  '.join(self._ifupdownd)))
        if self._unknown:
            _stderr(
                "Found some files we do not know about, check manually:\n"
                "  {}\n\n".format('\n  '.join(self._unknown)))


class InterfacesParser(object):
    "Interfaces file parser"

    def __init__(self, interfaces_lines):
        self._data = interfaces_lines
        self._sanitize()
        self._parse()
        self._join_inet_inet6()

    def _sanitize(self):
        newdata = []
        space_re = re.compile(r'\s+')
        for lineobj in self._data:
            line = lineobj.line.rstrip()

            # Blank line or only comment? Skip.
            if not line or line.lstrip().startswith('#'):
                continue

            # "[[:blank:]]#.*"? Drop trailing comment. Keep if the '#' is
            # not preceded by a space.
            if '#' in line and line.split('#', 1)[0].endswith((' ', '\t')):
                line = line.split('#', 1)[0].rstrip()

            # Collapse multiple leading blanks into one.
            if line.startswith((' ', '\t')):
                line = ' {}'.format(line.lstrip())

            # Collapse multiple blanks into one everywhere.
            line = space_re.sub(' ', line)

            # Store sanitized line.
            lineobj.line = line
            newdata.append(lineobj)

        self._data = newdata

    def _parse(self):
        context = None
        autos = set()
        config = {}

        try:
            for lineobj in self._data:
                if lineobj.line.startswith('auto '):
                    values = lineobj.line.split()  # auto lo eth0 eth1
                    for value in values[1:]:
                        autos.add(value)
                    context = None
                elif lineobj.line.startswith('iface '):
                    context = self._parse_iface(config, lineobj)
                elif context is not None and lineobj.line.startswith(' '):
                    self._parse_context(context, config[context], lineobj)
                else:
                    raise ValueError('unexpected/unimplemented')
        except Exception as e:  # assert/index/type/value
            raise ValueError('{}: parse fail at {!r}: {}'.format(
                lineobj.location, lineobj.line, ': '.join(
                    str(i) for i in e.args)))

        config_keys = set([i[0] for i in config.keys()])
        if config_keys != autos:
            raise ValueError(
                'not all interfaces are auto-on: {!r} != {!r}'.format(
                    sorted(config_keys), sorted(autos)))

        self._parsed = config

    def _parse_iface(self, config, lineobj):
        values = lineobj.line.split()  # iface eth0 inet{,6} static|...
        context = (values[1], values[2])  # (lo|eth0, inet|inet6)
        if context in config:
            raise ValueError('duplicate interface+protocol {!r}+{!r}'.format(
                context[0], context[1]))

        assert ':' not in values[1], 'iface:alias not available in netplan'
        assert values[2] in ('inet', 'inet6'), 'expected inet or inet6'
        type_ = values[3]  # auto|static|dhcp|loopback

        if context == ('lo', 'inet'):
            assert type_ == 'loopback', lineobj.line
            config[context] = {}
        elif type_ in 'static':
            config[context] = {}
        elif type_ == 'dhcp' and values[2] == 'inet':
            config[context] = {'dhcp4': 'yes'}
        elif type_ == 'auto' and values[2] == 'inet6':
            config[context] = {'dhcp6': 'yes'}
        else:
            raise ValueError('unexpected interface+protocol+type')

        return context

    def _parse_context(self, context, config, lineobj):
        cols = lineobj.line.split()
        cmd = cols.pop(0)
        if cmd == 'address':
            assert 'addresses' not in config, config
            assert len(cols) == 1, 'expected 1 argument to address'
            config['addresses'] = [cols[0]]
        elif cmd == 'netmask':
            assert 'addresses' in config, config
            assert '/' not in config['addresses'][0], config
            assert len(cols) == 1,  'expected 1 argument to netmask'
            bits = _netmaskbits(cols[0])
            config['addresses'][0] += '/{}'.format(bits)
        elif cmd == 'gateway' and context[1] == 'inet':
            assert 'gateway4' not in config, config
            assert len(cols) == 1, 'expected 1 argument to gateway'
            config['gateway4'] = cols[0]
        elif cmd == 'gateway' and context[1] == 'inet6':
            assert 'gateway6' not in config, config
            assert len(cols) == 1, 'expected 1 argument to gateway'
            config['gateway6'] = cols[0]
        elif cmd in ('network', 'broadcast'):
            _stderr(
                '{}: Ignoring {!r} of {} ({}). '
                'Your netmask is valid, right?\n'.format(
                    lineobj.location, cmd, context[0], context[1]))
        elif cmd == 'dns-nameservers':
            if 'nameservers' not in config:
                config['nameservers'] = {}
            assert 'addresses' not in config['nameservers'], config
            assert cols, 'expected one or more dns-nameservers arguments'
            config['nameservers']['addresses'] = cols
        elif cmd == 'dns-search':
            if 'nameservers' not in config:
                config['nameservers'] = {}
            assert 'search' not in config['nameservers'], config
            assert cols, 'expected one or more dns-search arguments'
            config['nameservers']['search'] = cols
        elif (len(cols) in (6, 8) and cols[0] in ('route', '/sbin/route') and
                ((cmd in 'up' and cols[1] == 'add') or
                 (cmd == 'down' and cols[1] == 'del')) and
                cols[2] in ('-host', '-net') and
                ((len(cols) == 6 and cols[4] == 'gw') or
                 (len(cols) == 8 and cols[4] == 'netmask' and
                  cols[6] == 'gw'))):
            # up route add -net NET (netmask MASK) gw GW
            # down route del -net NET (netmask MASK) gw GW
            if 'routes' not in config:
                config['routes'] = []

            if len(cols) == 6:
                if cols[2] == '-host':
                    assert '/' not in cols[3], cols
                    to = '{}/32'.format(cols[3])
                else:
                    assert '/' in cols[3], cols
                    to = cols[3]
                via = cols[5]
            else:
                assert cols[2] == '-net', cols
                assert '/' not in cols[3], cols
                to = '{}/{}'.format(cols[3], _netmaskbits(cols[5]))
                via = cols[7]

            route = {'to': to, 'via': via}

            if cmd == 'up':
                config['routes'].append(route)
            elif cmd == 'down':
                assert route in config['routes'], config['routes']
        else:
            raise ValueError('cmd {!r} unknown/unimplemented'.format(cmd))

    def _join_inet_inet6(self):
        ethernets = {}
        for (iface, proto), values in self._parsed.items():
            if iface not in ethernets:
                ethernets[iface] = {}
            for key, value in values.items():
                if isinstance(value, list):
                    if key not in ethernets[iface]:
                        ethernets[iface][key] = []
                    ethernets[iface][key].extend(value)
                else:
                    assert key not in ethernets[iface], (ethernets, iface, key)
                    ethernets[iface][key] = value

        # We should add it if we want extra addresses, but generally we don't.
        assert 'lo' in ethernets, ethernets
        assert ethernets['lo'] == {}
        del ethernets['lo']

        self._ethernets = ethernets

    def _to_yaml(self, value, indent=''):
        if isinstance(value, str):
            if ':' in value:
                assert '"' not in value, value
                return '"{}"'.format(value)
            return value

        ret = []
        if isinstance(value, dict):
            key_value = value.items()
            if not isinstance(value, OrderedDict):
                key_value = sorted(key_value)
            for key, value in key_value:
                inside = self._to_yaml(value, indent + '  ')
                if isinstance(inside, list):
                    ret.append('{}{}:'.format(indent, key))
                    ret.extend(inside)
                else:
                    ret.append('{}{}: {}'.format(indent, key, inside))
        elif isinstance(value, list):
            for v in value:
                if (isinstance(v, dict) and
                        all(isinstance(k, str) for k in v.keys()) and
                        all(isinstance(v2, str) for v2 in v.values())):
                    kv = list(sorted(v.items()))
                    assert kv, 'empty dict unexpected'
                    for idx, (key, value) in enumerate(kv):
                        ret.append('{}{}{}: {}'.format(
                            indent, ('- ' if idx == 0 else '  '),
                            self._to_yaml(key), self._to_yaml(value)))
                else:
                    assert isinstance(v, str), v
                    ret.append('{}- {}'.format(indent, self._to_yaml(v)))
        else:
            assert False, value

        return ret

    def to_netplan(self):
        ret = {
            'network': OrderedDict([
                ('version', '2'),
                ('renderer', 'networkd'),
                ('ethernets', self._ethernets),
            ]),
        }
        return '\n'.join(self._to_yaml(ret)) + '\n'


def main():
    files = NetworkFiles()
    files.hint()
    files.warn()
    files.show()
    sys.exit(files.exitcode())


if __name__ == '__main__':
    main()
# vim: set ts=8 sw=4 sts=4 et ai: