gonets/SRV/services/network_service.py

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"""

    def __init__(self):
        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 для устройства"""
        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) для устройства"""
        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:
            # print(f"Ошибка при получении соединения для устройства {await device.interface}: {e}")
            # срёт ошибки дл отключенных интерфейсов
            # добавить потом условия для отключенных интерфейсов
            return "unknown"

    async def get_ip_config(self, device: NetworkDeviceGeneric,
                            ip_version: str) -> List[IndividualIPConfig]:
        """Получает конфигурацию 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 устройстве"""
        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]]:
        """Получает список сетевых интерфейсов"""
        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"""
        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]:
        """
        Получает список интерфейсов по состоянию
        """
        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:
        """Получает объект, содержащий список интерфейсов и координаты контейнера"""
        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:
        """Получает контейнер с сетевыми интерфейсами и координатами контейнера"""
        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]:
        """Получает список активных сетевых интерфейсов"""
        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]:
        """Получает список неактивных сетевых интерфейсов"""
        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:
        """Получает объект, содержащий список активных интерфейсов и координаты контейнера"""
        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:
        """Получает объект, содержащий список неактивных интерфейсов и координаты контейнера"""
        interfaces = await self.get_deactivated_interfaces(initial_positions)
        return NetworkInterfacesContainer(interfaces=interfaces,
                                          container_x=container_x,
                                          container_y=container_y)