371 lines
No EOL
17 KiB
Python
371 lines
No EOL
17 KiB
Python
from typing import List, Dict, Tuple, Union, Optional
|
||
from sdbus_async.networkmanager import (
|
||
NetworkManager,
|
||
NetworkDeviceGeneric,
|
||
IPv4Config,
|
||
IPv6Config,
|
||
NetworkManagerSettings,
|
||
NetworkDeviceWireless,
|
||
AccessPoint,
|
||
DeviceType,
|
||
)
|
||
from sdbus import sd_bus_open_system
|
||
from models.network import (
|
||
NetworkInterface,
|
||
IPv4Route,
|
||
IPv6Route,
|
||
IPConfig,
|
||
IndividualIPConfig,
|
||
NetworkInterfacesContainer,
|
||
WiFiDevice,
|
||
)
|
||
import subprocess
|
||
|
||
|
||
class NetworkService:
|
||
"""Сервис для взаимодействия с NetworkManager через D-Bus.
|
||
|
||
Предоставляет асинхронные методы для получения информации о сетевых интерфейсах,
|
||
их конфигурациях IP, Wi-Fi и состояниях.
|
||
"""
|
||
|
||
def __init__(self):
|
||
"""Инициализирует сервис NetworkManager.
|
||
|
||
Устанавливает системный D-Bus и создаёт объекты для взаимодействия с
|
||
NetworkManager и его настройками.
|
||
"""
|
||
self.system_bus = sd_bus_open_system()
|
||
self.nm = NetworkManager(self.system_bus)
|
||
self.nm_settings = NetworkManagerSettings(self.system_bus)
|
||
|
||
async def get_ip_method(self, device: NetworkDeviceGeneric, ip_type: str) -> str:
|
||
"""Получает метод IP (например, "auto", "manual") для указанного устройства.
|
||
|
||
Args:
|
||
device (NetworkDeviceGeneric): Объект устройства NetworkManager.
|
||
ip_type (str): Тип IP (например, "ipv4" или "ipv6").
|
||
|
||
Returns:
|
||
str: Метод IP или "unknown" в случае ошибки.
|
||
|
||
Raises:
|
||
Exception: Если произошла ошибка при получении настроек соединения.
|
||
"""
|
||
try:
|
||
config = await device.get_applied_connection()
|
||
settings = config[0].get(ip_type, {})
|
||
return str(settings.get("method", ("s", "unknown"))[1])
|
||
except Exception:
|
||
return "unknown"
|
||
|
||
async def get_connection_name(self, device: NetworkDeviceGeneric) -> str:
|
||
"""Извлекает имя соединения (GENERAL.CONNECTION) для устройства.
|
||
|
||
Args:
|
||
device (NetworkDeviceGeneric): Объект устройства NetworkManager.
|
||
|
||
Returns:
|
||
str: Имя соединения или "unknown" в случае ошибки.
|
||
|
||
Raises:
|
||
Exception: Если произошла ошибка при получении настроек соединения.
|
||
"""
|
||
try:
|
||
applied_connection = await device.get_applied_connection()
|
||
connection_settings = applied_connection[0]
|
||
connection_info = connection_settings.get("connection", {})
|
||
return str(connection_info.get("id", ("s", "unknown"))[1])
|
||
except Exception as e:
|
||
return "unknown"
|
||
|
||
async def get_ip_config(self, device: NetworkDeviceGeneric, ip_version: str) -> List[IndividualIPConfig]:
|
||
"""Получает конфигурацию IP для устройства.
|
||
|
||
Args:
|
||
device (NetworkDeviceGeneric): Объект устройства NetworkManager.
|
||
ip_version (str): Версия IP ("4" или "6").
|
||
|
||
Returns:
|
||
List[IndividualIPConfig]: Список конфигураций IP, включая адреса и маршруты.
|
||
|
||
Raises:
|
||
Exception: Если произошла ошибка при получении данных IP.
|
||
"""
|
||
configs = []
|
||
device_ip = await getattr(device, f"ip{ip_version}_config")
|
||
if device_ip != "/":
|
||
ip_config = globals()[f"IPv{ip_version}Config"](device_ip, self.system_bus)
|
||
address_data = await ip_config.address_data
|
||
for address in address_data:
|
||
config = IndividualIPConfig(
|
||
address=str(address["address"][1]),
|
||
routes=[
|
||
globals()[f"IPv{ip_version}Route"](
|
||
destination=route.get("dest", ("s", "0.0.0.0" if ip_version == "4" else "::"))[1],
|
||
prefix=route.get("prefix", ("u", 0))[1],
|
||
next_hop=route.get("next-hop", ("s", "0.0.0.0" if ip_version == "4" else "::"))[1],
|
||
metric=route.get("metric", ("u", 0))[1],
|
||
) for route in await ip_config.route_data
|
||
],
|
||
method=await self.get_ip_method(device, f"ipv{ip_version}"),
|
||
gateway=str(await ip_config.gateway),
|
||
dns=[
|
||
str(ns["address"][1]) if ip_version == "4" else str(ns)
|
||
for ns in await getattr(
|
||
ip_config,
|
||
"nameserver_data" if ip_version == "4" else "nameservers",
|
||
)
|
||
],
|
||
)
|
||
configs.append(config)
|
||
return configs
|
||
|
||
async def get_wifi_info(self, wifi_device: NetworkDeviceWireless) -> Dict[str, Optional[Union[str, int]]]:
|
||
"""Получает информацию о Wi-Fi устройстве.
|
||
|
||
Args:
|
||
wifi_device (NetworkDeviceWireless): Объект Wi-Fi устройства NetworkManager.
|
||
|
||
Returns:
|
||
Dict[str, Optional[Union[str, int]]]: Словарь с данными Wi-Fi (SSID, BSSID,
|
||
bitrate, strength, frequency, mode), где значения могут быть None.
|
||
|
||
Raises:
|
||
Exception: Если произошла ошибка при получении данных точки доступа.
|
||
"""
|
||
active_ap = await wifi_device.active_access_point
|
||
wifi_info = {
|
||
"ssid": None,
|
||
"bssid": None,
|
||
"bitrate": await wifi_device.bitrate,
|
||
"strength": None,
|
||
"frequency": None,
|
||
"mode": None,
|
||
}
|
||
if active_ap != "/":
|
||
ap = AccessPoint(active_ap, self.system_bus)
|
||
wifi_info.update({
|
||
"ssid": await ap.ssid,
|
||
"bssid": await ap.hw_address,
|
||
"strength": await ap.strength,
|
||
"frequency": await ap.frequency,
|
||
"mode": await ap.mode,
|
||
})
|
||
return wifi_info
|
||
|
||
async def get_network_interfaces(
|
||
self,
|
||
initial_positions: Dict[str, Tuple[int, int]] = None
|
||
) -> List[Union[NetworkInterface, WiFiDevice]]:
|
||
"""Получает список сетевых интерфейсов.
|
||
|
||
Args:
|
||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||
координат (x, y) для интерфейсов. По умолчанию None (используются (100, 100)).
|
||
|
||
Returns:
|
||
List[Union[NetworkInterface, WiFiDevice]]: Список объектов интерфейсов,
|
||
включая Wi-Fi устройства.
|
||
|
||
Исключает интерфейсы типа LOOPBACK и неактивные состояния (state == "10").
|
||
"""
|
||
devices_paths = await self.nm.get_devices()
|
||
network_interfaces = []
|
||
initial_positions = initial_positions or {}
|
||
|
||
for device_path in devices_paths:
|
||
generic_device = NetworkDeviceGeneric(device_path, self.system_bus)
|
||
interface_name = str(await generic_device.interface)
|
||
device_type = DeviceType(await generic_device.device_type).name
|
||
|
||
if device_type == "LOOPBACK":
|
||
continue
|
||
|
||
device_state = str(await generic_device.state)
|
||
if device_state == "10":
|
||
continue
|
||
|
||
common_data = {
|
||
"name": interface_name,
|
||
"type": device_type,
|
||
"device_state": device_state,
|
||
"active_connection": await self.get_connection_name(generic_device),
|
||
"mac_address": str(await generic_device.hw_address),
|
||
"device_mtu": str(await generic_device.mtu),
|
||
"ipv4": IPConfig(configurations=await self.get_ip_config(generic_device, "4")),
|
||
"ipv6": IPConfig(configurations=await self.get_ip_config(generic_device, "6")),
|
||
"x": initial_positions.get(interface_name, (100, 100))[0],
|
||
"y": initial_positions.get(interface_name, (100, 100))[1],
|
||
}
|
||
|
||
if device_type == "WIFI":
|
||
wifi_device = NetworkDeviceWireless(device_path, self.system_bus)
|
||
wifi_info = await self.get_wifi_info(wifi_device)
|
||
network_interfaces.append(WiFiDevice(**common_data, **wifi_info))
|
||
else:
|
||
network_interfaces.append(NetworkInterface(**common_data))
|
||
|
||
return network_interfaces
|
||
|
||
async def get_hostname(self) -> Dict[str, str]:
|
||
"""Возвращает имя хоста из команды hostname в формате JSON.
|
||
|
||
Returns:
|
||
Dict[str, str]: Словарь с ключом "hostname" и значением имени хоста.
|
||
|
||
Raises:
|
||
RuntimeError: Если произошла ошибка при выполнении команды hostname
|
||
или обработке вывода.
|
||
"""
|
||
try:
|
||
result = subprocess.run(["hostname"],
|
||
capture_output=True,
|
||
text=True,
|
||
check=True)
|
||
hostname = result.stdout.strip()
|
||
return {"hostname": hostname}
|
||
except subprocess.CalledProcessError as e:
|
||
raise RuntimeError(f"Failed to execute hostname: {e}")
|
||
except Exception as e:
|
||
raise RuntimeError(f"Error processing hostname output: {e}")
|
||
|
||
async def get_interfaces_by_state(
|
||
self,
|
||
state: str,
|
||
initial_positions: Dict[str, Tuple[int, int]] = None
|
||
) -> List[NetworkInterface]:
|
||
"""Получает список интерфейсов по указанному состоянию.
|
||
|
||
Args:
|
||
state (str): Состояние интерфейса (например, "100" для активных).
|
||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||
координат (x, y) для интерфейсов. По умолчанию None.
|
||
|
||
Returns:
|
||
List[NetworkInterface]: Список интерфейсов с указанным состоянием.
|
||
"""
|
||
all_interfaces = await self.get_network_interfaces(initial_positions)
|
||
return [
|
||
interface for interface in all_interfaces
|
||
if interface.device_state == state
|
||
]
|
||
|
||
async def get_interfaces_container(
|
||
self,
|
||
state: Optional[str] = None,
|
||
initial_positions: Dict[str, Tuple[int, int]] = None,
|
||
container_x: int = 0,
|
||
container_y: int = 0,
|
||
) -> NetworkInterfacesContainer:
|
||
"""Получает объект, содержащий список интерфейсов и координаты контейнера.
|
||
|
||
Args:
|
||
state (Optional[str], optional): Состояние интерфейсов (например, "100").
|
||
По умолчанию None (все интерфейсы).
|
||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||
координат (x, y) для интерфейсов. По умолчанию None.
|
||
container_x (int, optional): Координата X контейнера. По умолчанию 0.
|
||
container_y (int, optional): Координата Y контейнера. По умолчанию 0.
|
||
|
||
Returns:
|
||
NetworkInterfacesContainer: Объект с интерфейсами и координатами контейнера.
|
||
"""
|
||
if state:
|
||
interfaces = await self.get_interfaces_by_state(state, initial_positions)
|
||
else:
|
||
interfaces = await self.get_network_interfaces(initial_positions)
|
||
return NetworkInterfacesContainer(interfaces=interfaces,
|
||
container_x=container_x,
|
||
container_y=container_y)
|
||
|
||
async def get_network_interfaces_container(
|
||
self,
|
||
initial_positions: Optional[Dict[str, Tuple[int, int]]] = None,
|
||
) -> NetworkInterfacesContainer:
|
||
"""Получает контейнер с сетевыми интерфейсами и координатами контейнера.
|
||
|
||
Args:
|
||
initial_positions (Optional[Dict[str, Tuple[int, int]]], optional):
|
||
Словарь начальных координат (x, y) для интерфейсов. По умолчанию None.
|
||
|
||
Returns:
|
||
NetworkInterfacesContainer: Объект с интерфейсами и координатами (0, 0).
|
||
"""
|
||
interfaces = await self.get_network_interfaces(initial_positions)
|
||
return NetworkInterfacesContainer(interfaces=interfaces,
|
||
container_x=0,
|
||
container_y=0)
|
||
|
||
async def get_activated_interfaces(
|
||
self,
|
||
initial_positions: Dict[str, Tuple[int, int]] = None
|
||
) -> List[NetworkInterface]:
|
||
"""Получает список активных сетевых интерфейсов.
|
||
|
||
Args:
|
||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||
координат (x, y) для интерфейсов. По умолчанию None.
|
||
|
||
Returns:
|
||
List[NetworkInterface]: Список активных интерфейсов (state == "100").
|
||
"""
|
||
return await self.get_interfaces_by_state("100", initial_positions)
|
||
|
||
async def get_deactivated_interfaces(
|
||
self,
|
||
initial_positions: Dict[str, Tuple[int, int]] = None
|
||
) -> List[NetworkInterface]:
|
||
"""Получает список неактивных сетевых интерфейсов.
|
||
|
||
Args:
|
||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||
координат (x, y) для интерфейсов. По умолчанию None.
|
||
|
||
Returns:
|
||
List[NetworkInterface]: Список интерфейсов с state != "100".
|
||
"""
|
||
all_interfaces = await self.get_network_interfaces(initial_positions)
|
||
return [
|
||
interface for interface in all_interfaces
|
||
if interface.device_state != "100"
|
||
]
|
||
|
||
async def get_activated_interfaces_container(
|
||
self,
|
||
initial_positions: Dict[str, Tuple[int, int]] = None,
|
||
container_x: int = 0,
|
||
container_y: int = 0,
|
||
) -> NetworkInterfacesContainer:
|
||
"""Получает объект, содержащий список активных интерфейсов и координаты контейнера.
|
||
|
||
Args:
|
||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||
координат (x, y) для интерфейсов. По умолчанию None.
|
||
container_x (int, optional): Координата X контейнера. По умолчанию 0.
|
||
container_y (int, optional): Координата Y контейнера. По умолчанию 0.
|
||
|
||
Returns:
|
||
NetworkInterfacesContainer: Объект с активными интерфейсами и координатами.
|
||
"""
|
||
return await self.get_interfaces_container("100", initial_positions, container_x, container_y)
|
||
|
||
async def get_deactivated_interfaces_container(
|
||
self,
|
||
initial_positions: Dict[str, Tuple[int, int]] = None,
|
||
container_x: int = 0,
|
||
container_y: int = 0,
|
||
) -> NetworkInterfacesContainer:
|
||
"""Получает объект, содержащий список неактивных интерфейсов и координаты контейнера.
|
||
|
||
Args:
|
||
initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных
|
||
координат (x, y) для интерфейсов. По умолчанию None.
|
||
container_x (int, optional): Координата X контейнера. По умолчанию 0.
|
||
container_y (int, optional): Координата Y контейнера. По умолчанию 0.
|
||
|
||
Returns:
|
||
NetworkInterfacesContainer: Объект с неактивными интерфейсами и координатами.
|
||
"""
|
||
interfaces = await self.get_deactivated_interfaces(initial_positions)
|
||
return NetworkInterfacesContainer(interfaces=interfaces, container_x=container_x, container_y=container_y) |