gonets/SRV/services/ip_service.py

"""
Сервис проверки и применения IP-настроек через NetworkManager (D-Bus).

Назначение:

- Перед применением конфигурации проверяет занятость IP адреса, сканируя сегмент сети средствами ARP (через `NetworkScanner`).
- Применяет изменения профиля подключения через `sdbus_async.networkmanager` и выполняет переподключение интерфейса.

Требования/ограничения:

- Доступ к системной шине D-Bus и права на изменения сетевых профилей.
- Структура входных данных должна соответствовать схеме `ipv4_updates` для корректного обновления вложенных полей у профиля.
"""
from typing import Dict, Any
from sdbus_async.networkmanager import (
    NetworkManager,
    NetworkManagerSettings,
    NetworkConnectionSettings,
    NetworkDeviceGeneric,
)
import sdbus
from sdbus_async.networkmanager.settings import ConnectionProfile
from services.arp_service import NetworkScanner
from utils.i18n import get_translations
from babel.support import Translations
from fastapi import Depends

class IPService:
    """Сервис проверки доступности IP-адреса и применения настроек сети.

    Предоставляет асинхронные методы для проверки занятости IP-адреса с использованием
    ARP-сканирования и применения настроек через NetworkManager.
    """

    def __init__(self):
        """Инициализирует сервис IP.

        Создаёт объект ARP-сканера и устанавливает системную шину D-Bus как
        по умолчанию.
        """
        self.arp_scanner = NetworkScanner()
        self.bus = sdbus.sd_bus_open_system()
        sdbus.set_default_bus(self.bus)

    def _reset_bus(self):
        """Перезапускает соединение с системной шиной D-Bus.

        Закрывает текущее соединение и создаёт новое, устанавливая его как
        по умолчанию.
        """
        new_bus = sdbus.sd_bus_open_system()
        sdbus.set_default_bus(new_bus)
        if hasattr(self, "bus") and self.bus is not None:
            self.bus = None
        self.bus = new_bus

    async def check_ip_config(self, args: Dict[str, Any], translations: Translations = Depends(get_translations())) -> Dict[str, Any]:
        """Проверяет доступность IP-адреса перед его применением.

        Выполняет ARP-сканирование сети для проверки, занят ли указанный IP-адрес.
        Перезапускает шину D-Bus перед выполнением.

        Args:
            args (Dict[str, Any]): Словарь с данными, включая "ipv4_updates"
                с полем "ipv4.address-data" для нового IP-адреса.
            translations (Translations): Объект переводов.

        Returns:
            Dict[str, Any]: Словарь с ключами:
                - "available" (bool): Доступен ли IP-адрес.
                - "message" (str): Сообщение о результате проверки.

        Raises:
            Exception: Если произошла ошибка при сканировании ARP.
        """
        self._reset_bus()
        clients = self.arp_scanner.scan_network()
        new_ip = args["ipv4_updates"]["ipv4.address-data"][0]["address"]
        if any(client["ip"] == new_ip for client in clients):
            return {
                "available": False,
                "message": translations.gettext("IP-адрес {ip} уже занят.").format(ip=new_ip)
            }
        return {
            "available": True,
            "message": translations.gettext("Настройки сетевого профиля доступны для использования.")
        }

    async def apply_ip_config(self, args: Dict[str, Any], translations: Translations = Depends(get_translations())) -> None:
        """Применяет проверенные настройки IP к указанному соединению.

        Перезапускает шину D-Bus, обновляет настройки соединения через NetworkManager,
        отключает и переподключает устройство к сети.

        Args:
            args (Dict[str, Any]): Словарь с данными, включая:
                - "connection_id" (str): Идентификатор соединения.
                - "ipv4_updates" (Dict): Обновления настроек IPv4.
            translations (Translations): Объект переводов.

        Raises:
            ValueError: Если соединение с указанным ID не найдено.
            RuntimeError: Если устройство для интерфейса не найдено.

        Note:
            Требуются права для изменения настроек через NetworkManager.
        """
        self._reset_bus()
        nm = NetworkManager()
        settings = NetworkManagerSettings()
        connection_paths = await settings.get_connections_by_id(
            args["connection_id"])

        if not connection_paths:
            raise ValueError(translations.gettext("Подключение {connection_id} не найдено").format(connection_id=args["connection_id"]))

        connection = NetworkConnectionSettings(connection_paths[0])
        profile = await connection.get_profile()
        settings_dict = profile.to_settings_dict()

        for key_path, value in args["ipv4_updates"].items():
            section, *props = key_path.split(".")
            current = settings_dict[section]
            for prop in props[:-1]:
                current = current.setdefault(prop, {})
            current[props[-1]] = value

        new_profile = ConnectionProfile.from_settings_dict(settings_dict)
        await connection.update(new_profile.to_dbus())
        print(f"Updated settings for {args['connection_id']}")

        interface_name = settings_dict["connection"]["interface-name"]
        devices = await nm.get_devices()
        for device_path in devices:
            device = NetworkDeviceGeneric(device_path)
            if await device.interface == interface_name:
                await device.disconnect()
                print(f"Disconnected from {interface_name}")
                await nm.activate_connection(connection=connection_paths[0], device=device_path)
                print(f"Reconnected to {args['connection_id']}")
                return

        raise RuntimeError(translations.gettext("Устройство для интерфейса {interface} не найдено").format(interface=interface_name))