pybatmesh/naxalnet/network.py

# This file is part of naxalnet.
# Copyright (C) 2021 The naxalnet Authors

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.

# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""
network.py
----------

This submodule manages the systemd-networkd configuration. This is used to
add configuration files to the systemd-networkd runtime directory.

Some configuration files have variables which should be substituted by
str.format() in python. The values for these variables can be set using
NetworkD.set_vars(). See files in the systemd-networkd directory for
examples.
"""

import subprocess
from pathlib import Path
from gi.repository import GObject
from dasbus.connection import SystemMessageBus
from dasbus.loop import EventLoop


NETWORKD_BUS = "org.freedesktop.network1"
NETWORKD_PATH = "/org/freedesktop/network1"


class NetworkD:
    """
    Control systemd-networkd using configuration files. Since these
    were made for use by naxalnet only, the class is not suitable for
    importing outside naxalnet.
    """

    def __init__(self, runtime_dir="/run/systemd/network", bus=SystemMessageBus()):
        self._bus = bus
        self.proxy_reload()

        self.variables = {}
        self.runtime_path = Path(runtime_dir)
        # Create the runtime directory if it doesn't exist
        self.runtime_path.mkdir(parents=True, exist_ok=True)

    def set_vars(self, **variables):
        """set the variables to replace with str.format"""
        self.variables = variables

    def proxy_reload(self) -> None:
        """reload the proxy"""
        self.proxy = self._bus.get_proxy(NETWORKD_BUS, NETWORKD_PATH)

    def reload(self) -> None:
        """
        Reload the systemd-networkd configuration. This is used by many
        class methods after doing their job.
        """
        self.proxy.Reload()

    def add_config(self, name: str) -> None:
        """add config file to runtime directory and reload networkd"""
        source = Path(name)
        destination = self.runtime_path / source.name

        # Substitute variables in the config
        text = source.read_text(encoding="utf-8").format(**self.variables)
        # now write it to a runtime config of the same name
        destination.write_text(text, encoding="utf-8")
        self.reload()

    def is_routable(self) -> bool:
        """returns true if any interface is routable"""
        return self.proxy.AddressState == "routable"

    def delete_interface(self, name: str) -> None:
        """delete the given interface"""
        # If anyone knows a better way of doing this, create
        # an issue and get things done
        subprocess.run(["networkctl", "delete", name], check=True)
        # This is probably not required. This is mainly to shut up
        # pylint's messages
        self.reload()

    def disable_config(self, name: str) -> None:
        """
        Disable or mask the config of the same name. This can only be
        used for configs in /usr/lib/systemd/network and
        /usr/local/lib/systemd/network. It works on the same principle
        used by systemctl mask, that is, it created a symlink of the same
        name in the runtime directory and links it to /dev/null.
        """
        path = self.runtime_path / name
        path.symlink_to("/dev/null")
        self.reload()

    def remove_config(self, name: str) -> None:
        """
        remove the file called 'name' from the runtime dir and reload
        """
        path = self.runtime_path / name
        path.unlink()
        self.reload()

    def remove_all_configs(self) -> None:
        """
        Remove all configs in runtime_path. This will remove all files
        in runtime_path without checking who put them there.
        """
        for i in self.runtime_path.iterdir():
            self.remove_config(i.name)


class NetworkLoop(NetworkD):
    """Used to wait until a condition is met

    Available methods:

    NetworkLoop.wait_until_routable(timeout=0):
        return true when the network is routable, or false when timed out

    """

    def __init__(self, *args, **kwargs):
        # first, initialise the parent object
        super().__init__(*args, **kwargs)
        self.waitfor = None
        self.wait_function = None
        self.loop = EventLoop()

    def start_loop(self):
        """start the dasbus loop"""
        self.proxy.PropertiesChanged.connect(self.on_properties_changed)
        self.loop.run()

    def wait_until_routable(self, timeout=0):
        """
        Wait until timeout in milliseconds and returns True when any
        network interface is shown routable by networkd. Does not wait
        for timeout if timeout==0
        """
        self.wait_for_change("AddressState", self.on_addressstate_change)
        self.setup_timeout(timeout)
        return self.is_routable()

    def wait_for_change(self, name, function):
        """used by the public functions"""
        self.waitfor = name
        self.wait_function = function
        self.start_loop()

    def on_addressstate_change(self):
        """quit the loop if the network is routable"""
        if self.is_routable():
            self.loop.quit()

    def on_properties_changed(self, bus_interface, data, blah):
        """give this function some documentation"""
        if self.waitfor in data:
            return self.wait_function()

    def setup_timeout(self, timeout):
        """setup a timeout"""
        if timeout != 0:
            GObject.timeout_add(timeout, self.on_timeout)

    def on_timeout(self):
        """called by dasbus when a timeout occurs"""
        self.loop.quit()