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
from utils.i18n import get_translations, _
from babel.support import Translations
from fastapi import Depends

class NetworkService:
    """Сервис для взаимодействия с NetworkManager через D-Bus.

    Предоставляет асинхронные методы для получения информации о сетевых интерфейсах,
    их конфигурациях IP, Wi-Fi и состояниях.
    """

    def __init__(self, translations: Translations = Depends(get_translations())):
        """Инициализирует сервис NetworkManager.

        Устанавливает системный D-Bus и создаёт объекты для взаимодействия с
        NetworkManager и его настройками.
        """
        self.system_bus = sd_bus_open_system()
        self.nm = NetworkManager(self.system_bus)
        self.nm_settings = NetworkManagerSettings(self.system_bus)
        self.translations = translations

    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(self.translations.gettext("Не удалось выполнить команду hostname: {error}").format(error=str(e)))
        except Exception as e:
            raise RuntimeError(self.translations.gettext("Ошибка обработки вывода hostname: {error}").format(error=str(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)