Fallback for func check ip if scapy/arp not working

SRV/services/ip_service.py

@@ -11,7 +11,10 @@
 - Доступ к системной шине D-Bus и права на изменения сетевых профилей.
 - Структура входных данных должна соответствовать схеме `ipv4_updates` для корректного обновления вложенных полей у профиля.
 """
-from typing import Dict, Any
+import asyncio
+import logging
+from typing import Dict, Any, Optional
+import subprocess
 from sdbus_async.networkmanager import (
     NetworkManager,
     NetworkManagerSettings,
@@ -25,6 +28,8 @@ from utils.i18n import _load_translations
 from babel.support import Translations
 from fastapi import Depends
 
+logger = logging.getLogger(__name__)
+
 class IPService:
     """Сервис проверки доступности IP-адреса и применения настроек сети.
 
@@ -75,8 +80,21 @@ class IPService:
         """
         translations = translations or _load_translations("ru")
         self._reset_bus()
-        clients = self.arp_scanner.scan_network()
         new_ip = args["ipv4_updates"]["ipv4.address-data"][0]["address"]
+        logger.info("check_ip_config: start ARP scan for %s", new_ip)
+
+        try:
+            clients = await asyncio.wait_for(
+                asyncio.to_thread(self.arp_scanner.scan_network),
+                timeout=10,
+            )
+        except asyncio.TimeoutError as exc:
+            logger.error("check_ip_config: ARP scan timed out for %s", new_ip)
+            return self._fallback_ping(new_ip, translations)
+        except Exception as exc:
+            logger.exception("check_ip_config: ARP scan failed for %s", new_ip)
+            return self._fallback_ping(new_ip, translations, error=str(exc))
+
         if any(client["ip"] == new_ip for client in clients):
             return {
                 "available": False,
@@ -87,6 +105,32 @@ class IPService:
             "message": translations.gettext("Настройки сетевого профиля доступны для использования.")
         }
 
+    def _fallback_ping(self, ip: str, translations: Translations, error: Optional[str] = None) -> Dict[str, Any]:
+        """Резервная проверка через ping, если ARP-скан недоступен.
+
+        Не блокирует применение: если ping не отвечает, считаем IP свободным,
+        но возвращаем предупреждение.
+        """
+        try:
+            logger.info("check_ip_config: fallback ping for %s after ARP failure (%s)", ip, error or "timeout")
+            result = subprocess.run(
+                ["ping", "-c", "1", "-W", "1", ip],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+            )
+            if result.returncode == 0:
+                return {
+                    "available": False,
+                    "message": translations.gettext("IP-адрес {ip} отвечает на ping").format(ip=ip),
+                }
+        except Exception as exc:
+            logger.warning("check_ip_config: ping fallback failed for %s: %s", ip, exc)
+
+        return {
+            "available": True,
+            "message": translations.gettext("Не удалось проверить ARP, IP {ip} считается свободным").format(ip=ip),
+        }
+
     async def apply_ip_config(self, args: Dict[str, Any], translations: Translations = None) -> None:
         """Применяет проверенные настройки IP к указанному соединению.
 
@@ -110,14 +154,32 @@ class IPService:
         self._reset_bus()
         nm = NetworkManager()
         settings = NetworkManagerSettings()
-        connection_paths = await settings.get_connections_by_id(
+
-            args["connection_id"])
+        logger.info("apply_ip_config: start for connection_id=%s", args.get("connection_id"))
+
+        try:
+            connection_paths = await asyncio.wait_for(
+                settings.get_connections_by_id(args["connection_id"]),
+                timeout=5,
+            )
+        except asyncio.TimeoutError as exc:
+            logger.error("apply_ip_config: timeout while fetching connection %s", args.get("connection_id"))
+            raise RuntimeError(
+                translations.gettext("NetworkManager не ответил при поиске профиля")
+            ) from exc
 
         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()
+        try:
+            profile = await asyncio.wait_for(connection.get_profile(), timeout=5)
+        except asyncio.TimeoutError as exc:
+            logger.error("apply_ip_config: timeout while reading profile %s", args.get("connection_id"))
+            raise RuntimeError(
+                translations.gettext("NetworkManager не ответил при чтении профиля")
+            ) from exc
+
         settings_dict = profile.to_settings_dict()
 
         for key_path, value in args["ipv4_updates"].items():
@@ -128,18 +190,51 @@ class IPService:
             current[props[-1]] = value
 
         new_profile = ConnectionProfile.from_settings_dict(settings_dict)
-        await connection.update(new_profile.to_dbus())
+        try:
-        print(f"Updated settings for {args['connection_id']}")
+            await asyncio.wait_for(connection.update(new_profile.to_dbus()), timeout=5)
+        except asyncio.TimeoutError as exc:
+            logger.error("apply_ip_config: timeout while updating profile %s", args.get("connection_id"))
+            raise RuntimeError(
+                translations.gettext("NetworkManager не ответил при обновлении профиля")
+            ) from exc
 
-        interface_name = settings_dict["connection"]["interface-name"]
+        logger.info("apply_ip_config: settings updated for %s", args.get("connection_id"))
-        devices = await nm.get_devices()
+
+        interface_name: Optional[str] = settings_dict["connection"].get("interface-name")
+
+        devices = await asyncio.wait_for(nm.get_devices(), timeout=5)
+        # Если interface-name не задан, пытаемся найти устройство по активному подключению
+        if not interface_name:
             for device_path in devices:
                 device = NetworkDeviceGeneric(device_path)
-            if await device.interface == interface_name:
+                try:
-                await device.disconnect()
+                    applied = await asyncio.wait_for(device.get_applied_connection(), timeout=3)
-                print(f"Disconnected from {interface_name}")
+                    connection_info = applied[0].get("connection", {})
-                await nm.activate_connection(connection=connection_paths[0], device=device_path)
+                    active_id = connection_info.get("id", ("s", None))[1]
-                print(f"Reconnected to {args['connection_id']}")
+                    if active_id == args["connection_id"]:
+                        interface_name = await asyncio.wait_for(device.interface, timeout=3)
+                        break
+                except Exception as exc:
+                    logger.debug("apply_ip_config: unable to derive interface from device %s: %s", device_path, exc)
+                    continue
+
+        if not interface_name:
+            raise RuntimeError(
+                translations.gettext("Не удалось определить интерфейс для профиля {connection_id}").format(
+                    connection_id=args["connection_id"]
+                )
+            )
+
+        for device_path in devices:
+            device = NetworkDeviceGeneric(device_path)
+            if await asyncio.wait_for(device.interface, timeout=3) == interface_name:
+                await asyncio.wait_for(device.disconnect(), timeout=5)
+                logger.info("apply_ip_config: disconnected from %s", interface_name)
+                await asyncio.wait_for(
+                    nm.activate_connection(connection=connection_paths[0], device=device_path),
+                    timeout=10,
+                )
+                logger.info("apply_ip_config: reconnected to %s", args.get("connection_id"))
                 return
 
         raise RuntimeError(translations.gettext("Устройство для интерфейса {interface} не найдено").format(interface=interface_name))