pybatmesh/naxalnet/iwd.py

#!/usr/bin/env python3

# 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/>.


"""
iwd.py
------

This file contains methods to communicate with iwd via
its D-Bus API and control WiFi adapters.

Some terms used here, such as device and adapter might
confuse you if you haven't used iwctl before.
Just as a quick reference, here is a list of terms
and what they mean:

- ad-hoc: a mode supported by some WiFi adapters to
          start a decentralised network, where there
          is no central point of failure.

- ap: a mode used to start a central access point
      so that other machines without naxalnet can
      connect to the mesh. AP is also known as
      WiFi hotspot.

- station: this is the mode most WiFi adapters use
           by default. This mode is used to connect to
           an ap. naxalnet DOES NOT use this mode.

- adapter: a physical WiFi chip or something similar
           that is present inside most laptops and phones
           or can be connected via USB to a machine.

- device: an interface provided by the kernel to control
          an adapter. Some adapters can have multiple
          devices so that you can start an ap on one device
          and an ad-hoc on the other. By default, iwd starts
          only one device each for one adapter.

- machine: Since iwd uses the term device for a
           WiFi interface, we use the word machine to
           refer to a computer, or a laptop, or a phone.

- node: a machine that runs naxalnet and is therefore
        connected to the mesh.
"""

from dasbus.connection import SystemMessageBus

IWD_BUS = "net.connman.iwd"
IWD_ROOT_PATH = "/"
IWD_DEVICE_INTERFACE = "net.connman.iwd.Device"
IWD_ADAPTER_INTERFACE = "net.connman.iwd.Adapter"

# If you are new to D-Bus, you might want to use a program
# such as D-Feet (https://wiki.gnome.org/Apps/DFeet) for reference.
# And try out iwctl to understand iwd's bus objects.


class IWD:
    """Manage iwd via dbus"""

    def __init__(self, bus=SystemMessageBus()):
        # self._bus and self._proxy are meant for use only in this file
        self._bus = bus
        self.reload()

    def reload(self):
        """reload the proxy"""
        self._proxy = self._bus.get_proxy(IWD_BUS, IWD_ROOT_PATH)

    def get_name_from_path(self, path: str) -> str:
        """
        returns device or adapter name when d-bus path is given as arg
        """
        proxy = self._bus.get_proxy(IWD_BUS, path)
        return proxy.Name

    def get_device_path_from_name(self, name: str) -> str:
        """returns path of device as str"""
        device_paths = self.get_all_device_paths()
        for i in device_paths:
            proxy = self._bus.get_proxy(IWD_BUS, i)
            if proxy.Name == name:
                return i
            # If no devices were found, return None
        return None

    def get_adapter_path_from_name(self, name: str) -> str:
        """returns path of adapter as str"""
        adapter_paths = self.get_all_adapter_paths()
        for i in adapter_paths:
            proxy = self._bus.get_proxy(IWD_BUS, i)
            if proxy.Name == name:
                return i
            # If no adapters were found
        return None

    def get_all_device_paths(self) -> list:
        """returns list of paths of all devices"""
        objects = self._proxy.GetManagedObjects()
        paths = []
        for key, value in objects.items():
            # if value is a device, add its path to paths
            if IWD_DEVICE_INTERFACE in value:
                paths.append(key)
        return paths

    def get_all_adapter_paths(self) -> list:
        """returns list of paths of all adapters"""
        objects = self._proxy.GetManagedObjects()
        paths = []

        for key, value in objects.items():
            # if value is an adapter, add its path to paths
            if IWD_ADAPTER_INTERFACE in value:
                paths.append(key)
        return paths

    def get_devices(self) -> list:
        """
        returns list of device names as str
        example: ["wlan0", "wlan1"]
        """
        devices = []
        device_paths = self.get_all_device_paths()

        for i in device_paths:
            name = self.get_name_from_path(i)
            devices.append(name)
        return devices

    def get_adapters(self) -> list:
        """
        returns list of adapters
        example: ["phy0","phy1"]
        """
        adapters = []
        adapter_paths = self.get_all_adapter_paths()

        for i in adapter_paths:
            name = self.get_name_from_path(i)
            adapters.append(name)
        return adapters


class Device:
    """
    control devices with iwd
    name: name of device (str)
    adapter: name of adapter (str)
    """

    def __init__(self, name: str, bus=SystemMessageBus()):
        self._iwd = IWD(bus)
        self._bus = self._iwd._bus
        self._path = self._iwd.get_device_path_from_name(name)
        self.reload()

    def __str__(self):
        return self.name

    def is_powered_on(self) -> bool:
        """returns True if devie is powered on"""
        return self._proxy.Powered

    def power_on(self):
        """Turn on the device and reload the proxy"""
        self._proxy.Powered = True
        self.reload()

    def power_off(self):
        """Turn off the device and reload the proxy"""
        self._proxy.Powered = False
        self.reload()

    def reload(self):
        """reload the proxy after changing mode"""
        self._proxy = self._bus.get_proxy(IWD_BUS, self._path)
        self.name = self._proxy.Name
        adapter_path = self._proxy.Adapter
        # name of adapter ('phy0' for example)
        self.adapter = self._iwd.get_name_from_path(adapter_path)

    def is_adhoc_started(self) -> bool:
        """
        Returns True if an adhoc network is started on this device.
        Returns False if the network is in staring stage, device
        is not powered on or not in ad-hoc mode.
        """
        if self.is_powered_on() and self.get_mode() == "ad-hoc":
            return self._proxy.Started
        # If above condition is not true, return False
        return False

    def is_ap_started(self) -> bool:
        """
        Same as is_adhoc_started(), but for ap
        """
        if self.is_powered_on() and self.get_mode() == "ap":
            return self._proxy.Started
        return False

    def get_mode(self) -> str:
        """
        returns the mode in which the device is in
        example: "ap"
        """
        return self._proxy.Mode

    def set_mode(self, mode: str):
        """change the device mode to mode"""
        self._proxy.Mode = mode
        self.reload()

    def start_adhoc_open(self, name: str):
        """
        Create ad-hoc network with name, changing mode to ad-hoc
        if it isn't already on ad-hoc and power onn the device
        if it is off
        """
        if self.get_mode() != "ad-hoc":
            self.set_mode("ad-hoc")

        if not self.is_powered_on():
            self.power_on()

        # Stop adhoc if already started
        self.stop_adhoc()

        self._proxy.StartOpen(name)

    def stop_adhoc(self):
        """stop adhoc if adhoc is started"""
        if self.is_adhoc_started():
            self._proxy.Stop()
            self.reload()

    def start_ap(self, ssid, passwd):
        """
        Create ap network, changing mode to ap
        if it isn't already on ap and turning
        on the device if it is off
        """
        if self.get_mode() != "ap":
            self.set_mode("ap")

        if not self.is_powered_on():
            self.power_on()

        # Stop ap if already started
        self.stop_ap()

        self._proxy.Start(ssid, passwd)

    def stop_ap(self):
        """stop ap if an ap is started"""
        if self.is_ap_started():
            self._proxy.Stop()
            self.reload()


class Adapter:
    """represents an adapter as a python object"""

    def __init__(self, name: str, bus=SystemMessageBus()):
        self._iwd = IWD(bus)
        self._bus = self._iwd._bus
        self._path = self._iwd.get_adapter_path_from_name(name)
        # Initialise self._proxy
        self.reload()

    def __str__(self):
        return self.name

    def reload(self):
        """reload the proxy after changing mode"""
        self._proxy = self._bus.get_proxy(IWD_BUS, self._path)
        self.name = self._proxy.Name
        self.supported_modes = self._proxy.SupportedModes

    def is_powered_on(self) -> bool:
        """returns True if adapter is powered on, False otherwise"""
        return self._proxy.Powered

    def power_on(self):
        """power on the adapter"""
        self._proxy.Powered = True

    def power_off(self):
        """power off the adapter"""
        self._proxy.Powered = False

    def supports_mode(self, mode: str) -> bool:
        """
        Returns True if the adapter supports the mode.
        mode can be "ad-hoc", "ap" or "station"
        """
        return mode in self.supported_modes