Source code for services.arp_service

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

Назначение:

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

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

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

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

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



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

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

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

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

        Инициализирует интерфейс, информацию об IP-адресах и локальный MAC-адрес.
        """
        self.interface = (
            interface or netifaces.gateways()["default"][netifaces.AF_INET][1])
        self.ip_info = netifaces.ifaddresses(self.interface).get(
            netifaces.AF_INET, [])
        self.networks = [
            IPNetwork(f"{ip['addr']}/{ip['netmask']}") for ip in self.ip_info
        ]
        self.local_mac = netifaces.ifaddresses(
            self.interface)[netifaces.AF_LINK][0]["addr"]
        self.translations = translations

    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-адреса интерфейса и удаляет дубликаты.
        """
        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:
            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)