326 lines
10 KiB
Python
326 lines
10 KiB
Python
# This file is part of pybatmesh.
|
|
# Copyright (C) 2021 The pybatmesh 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 pybatmesh 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. pybatmesh 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 pybatmesh and is therefore connected to the
|
|
mesh.
|
|
"""
|
|
|
|
from dasbus.connection import SystemMessageBus
|
|
from pybatmesh.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 submodule
|
|
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:
|
|
# See comment in the function below
|
|
path = i
|
|
return path
|
|
|
|
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:
|
|
# We could have just used return here, but shutting up
|
|
# pylint has a greter priority at the moment
|
|
path = i
|
|
return path
|
|
|
|
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 all 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. Used liberally by other
|
|
members to work around errors
|
|
"""
|
|
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
|
|
"""
|
|
# Stop adhoc if already started
|
|
self.stop_adhoc()
|
|
|
|
if self.get_mode() != "ad-hoc":
|
|
self.set_mode("ad-hoc")
|
|
|
|
if not self.is_powered_on():
|
|
self.power_on()
|
|
|
|
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
|
|
"""
|
|
|
|
# Stop ap if already started
|
|
self.stop_ap()
|
|
|
|
if self.get_mode() != "ap":
|
|
self.set_mode("ap")
|
|
|
|
if not self.is_powered_on():
|
|
self.power_on()
|
|
|
|
logger.debug("Starting ap on %s with ssid %s", self.name, ssid)
|
|
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)
|
|
self.reload()
|
|
|
|
def power_off(self):
|
|
"""power off the adapter"""
|
|
self._proxy.Powered = False
|
|
logger.debug("Powered off adapter %s", self.name)
|
|
self.reload()
|
|
|
|
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
|