gonets/SRV/services/arp_service.py

"""
Сервис низкоуровневого ARP-сканирования локальной сети.

Назначение:

- Обнаружение активных узлов в пределах сетей хоста путём широковещательных ARP-запросов от имени выбранного интерфейса.
- Помощь сервисам более высокого уровня (например, IPService) при проверке занятости адресов перед применением конфигураций.

Особенности реализации:

- Определяет подключённые сети интерфейса через `netifaces` и строит CIDR для сканирования.
- Формирует ARP-пакеты с помощью `scapy` (Ether/ARP) и отправляет их с таймаутом.
- Добавляет локальный адрес интерфейса в результирующий список клиентов.

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

- Для отправки ARP-кадров могут потребоваться повышенные привилегии или capability на сырой сокет.
"""
import os
import logging
from fastapi import Depends
import netifaces
from netaddr import IPNetwork
from utils.i18n import get_translations, _
from babel.support import Translations

logger = logging.getLogger(__name__)

try:
    from scapy.layers.l2 import ARP, Ether
    from scapy.sendrecv import srp
except Exception as exc:  # pragma: no cover - fallback for doc builds
    if os.getenv("SPHINX_BUILD"):
        ARP = Ether = None  # type: ignore

        def srp(*args, **kwargs):
            raise RuntimeError("scapy is unavailable during documentation build") from exc
    else:
        raise

class NetworkScanner:
    """Инициализация сетевого сканера.

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

    def __init__(self, interface=None, translations: Translations = Depends(get_translations())):
        """Инициализация параметров сетевого сканера.

        Args:
            interface (str, optional): Название сетевого интерфейса. По умолчанию
                используется интерфейс по умолчанию из netifaces.gateways().
            translations: Объект переводов для текущего языка.

        Инициализирует интерфейс, информацию об IP-адресах и локальный MAC-адрес.
        """
        self.translations = translations
        self.interface = None
        self.ip_info = []
        self.networks = []
        self.local_mac = None

        try:
            candidate_interface = interface
            if not candidate_interface:
                gateways = netifaces.gateways()
                default_gateway = gateways.get("default", {}).get(netifaces.AF_INET)
                if default_gateway:
                    candidate_interface = default_gateway[1]
                else:
                    for iface in netifaces.interfaces():
                        if iface == "lo":
                            continue
                        addrs = netifaces.ifaddresses(iface)
                        if netifaces.AF_INET in addrs and addrs[netifaces.AF_INET]:
                            candidate_interface = iface
                            break

            if candidate_interface:
                self.interface = candidate_interface
                addresses = netifaces.ifaddresses(self.interface)
                self.ip_info = addresses.get(netifaces.AF_INET, [])
                self.networks = [
                    IPNetwork(
                        f"{ip['addr']}/{ip['netmask']}"
                    ) for ip in self.ip_info if ip.get("addr") and ip.get("netmask")
                ]
                link_info = addresses.get(netifaces.AF_LINK, [])
                if link_info:
                    self.local_mac = link_info[0].get("addr")
            else:
                logger.warning(
                    "NetworkScanner: IPv4 интерфейс не найден, сканирование будет недоступно"
                )
        except Exception as exc:
            if os.getenv("SPHINX_BUILD"):
                self.interface = interface or "eth0"
                self.ip_info = []
                self.networks = []
                self.local_mac = "00:00:00:00:00:00"
                logger.warning(
                    "NetworkScanner falling back to mocked interface during documentation build: %s",
                    exc,
                )
            else:
                logger.error("NetworkScanner инициализация завершилась ошибкой: %s", exc)
                raise

    def _create_arp_packet(self, target_ip):
        """Создаёт ARP-пакет для отправки на указанный IP-адрес.

        Args:
            target_ip (str): Целевой IP-адрес или подсеть для сканирования.

        Returns:
            scapy.packet.Packet: Сформированный ARP-пакет с Ethernet-заголовком.
        """
        arp = ARP(pdst=target_ip)
        ether = Ether(dst="ff:ff:ff:ff:ff:ff")
        return ether / arp

    def scan_network(self, attempts=3, timeout=2):
        """Сканирует сеть для обнаружения активных клиентов.

        Args:
            attempts (int, optional): Количество попыток отправки ARP-запросов.
                По умолчанию 3.
            timeout (int, optional): Таймаут ожидания ответа в секундах.
                По умолчанию 2.

        Returns:
            List[dict]: Список словарей с ключами "ip" и "mac", представляющих
                обнаруженных клиентов, включая локальные IP-адреса.

        Note:
            Учитывает локальные IP-адреса интерфейса и удаляет дубликаты.
        """
        if not self.interface or not self.networks or ARP is None or Ether is None:
            fallback_clients = []
            if self.ip_info:
                for ip_info in self.ip_info:
                    fallback_clients.append({
                        "ip": ip_info.get("addr"),
                        "mac": self.local_mac,
                    })
            return [client for client in fallback_clients if client["ip"]]

        all_clients = []

        for network in self.networks:
            target_ip = f"{network.network}/{network.prefixlen}"
            packet = self._create_arp_packet(target_ip)
            clients = []

            for _ in range(attempts):
                result = srp(packet,
                             timeout=timeout,
                             iface=self.interface,
                             verbose=0)[0]
                for _, received in result:
                    client = {"ip": received.psrc, "mac": received.hwsrc}
                    if client not in clients:
                        clients.append(client)

            all_clients.extend(clients)

        for ip_info in self.ip_info:
            if not ip_info.get("addr"):
                continue
            local_client = {
                "ip": ip_info["addr"],
                "mac": self.local_mac,
            }
            if local_client not in all_clients:
                all_clients.append(local_client)

        unique_clients = {tuple(client.items()) for client in all_clients}
        return [dict(client) for client in unique_clients]

    def print_clients(self, clients):
        """Выводит список адресов клиентов в формате таблицы.

        Args:
            clients (List[dict]): Список словарей с ключами "ip" и "mac",
                представляющих клиентов.

        Выводит IP-адреса и MAC-адреса в отсортированном по IP порядке.
        """
        print(self.translations.gettext("Доступные устройства в сети:"))
        print(self.translations.gettext("IP") + " " * 18 + self.translations.gettext("MAC"))
        for client in sorted(clients, key=lambda x: x["ip"]):
            print(f"{client['ip']:<16}    {client['mac']}")

# Оставил нативный запуск сервиса для тестов
if __name__ == "__main__":
    scanner = NetworkScanner()
    clients = scanner.scan_network()
    scanner.print_clients(clients)