pybatmesh/naxalnet/network.py
Pranav Jerry fc1f3d4216
another fix for #25
Or, to be more precise, a fix for the fix preceding the current fix for
issue #25 that just didn't fix things right. If your head spins, see the
previous commit, or call an ambulance (that is, if you are privileged
enough to access a phone and know your local helpline number)
2021-11-29 14:47:59 +05:30

195 lines
6.4 KiB
Python

# 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 dasbus.connection import SystemMessageBus
from dasbus.loop import EventLoop, GLib
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()):
print("NetworkD init")
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)
# TODO: remove all calls to print() before merging to master
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):
print("NetworkLoop init")
# 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"""
print("start loop")
print("waitfor", self.waitfor)
print("waitfor func", self.wait_function)
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
"""
print("wait until routable")
self.setup_timeout(timeout)
self.wait_for_change("AddressState", self.on_addressstate_change)
return self.is_routable()
def wait_for_change(self, name, function):
"""used by the public functions"""
print("wait for change")
self.waitfor = name
self.wait_function = function
self.start_loop()
def on_addressstate_change(self):
"""quit the loop if the network is routable"""
print("on addrstate change")
if self.is_routable():
self.loop.quit()
def on_properties_changed(self, bus_interface, data, blah):
"""give this function some documentation"""
print("on properties changed")
if self.waitfor in data:
return self.wait_function()
def setup_timeout(self, timeout):
"""setup a timeout"""
print("setup timeout")
if timeout != 0:
GLib.timeout_add(timeout, self.on_timeout)
def on_timeout(self):
"""called by dasbus when a timeout occurs"""
print("on timeout")
self.loop.quit()