pybatmesh/naxalnet/iwd.py

325 lines
10 KiB
Python

#!/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
from naxalnet.log import logger
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):
self._iwd = IWD()
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
logger.debug("Powered on %s", self.name)
self.reload()
def power_off(self):
"""Turn off the device and reload the proxy"""
self._proxy.Powered = False
logger.debug("Powered off %s", self.name)
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
logger.debug("Set mode on %s to %s", self.name, 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()
logger.debug("Starting ad-hoc on %s", self.name)
self._proxy.StartOpen(name)
def stop_adhoc(self):
"""stop adhoc if adhoc is started"""
if self.is_adhoc_started():
logger.debug("Stopping ad-hoc on %s", self.name)
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()
logger.debug(
"Starting ap on %s with ssid %s and password %s", self.name, ssid, passwd
)
self._proxy.Start(ssid, passwd)
def stop_ap(self):
"""stop ap if an ap is started"""
if self.is_ap_started():
logger.debug("Stopping ap on %s", self.name)
self._proxy.Stop()
self.reload()
class Adapter:
"""represents an adapter as a python object"""
def __init__(self, name: str):
self._iwd = IWD()
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
logger.debug("Powered on adapter %s", self.name)
def power_off(self):
"""power off the adapter"""
self._proxy.Powered = False
logger.debug("Powered off adapter %s", self.name)
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