from scapy.config import conf from mac_vendor_lookup import MacLookup from typing import List, Optional, Tuple, Dict import socket import logging import netifaces import ipaddress from concurrent.futures import ThreadPoolExecutor import time from functools import lru_cache import json from pathlib import Path from datetime import datetime, timedelta import subprocess from scapy.layers.l2 import ARP, Ether from scapy.sendrecv import srp from utils.i18n import get_translations from babel.support import Translations from fastapi import Depends from models.subnet import SubnetScanResult, Device logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) def get_subnet_service(translations: Translations = Depends(get_translations())): """Фабричная функция для создания экземпляра SubnetService. Использует зависимости FastAPI для получения объекта переводов и возвращает экземпляр SubnetService с текущей локализацией. Args: translations (Translations): Объект переводов для текущего языка. Returns: SubnetService: Экземпляр сервиса для работы с подсетями. """ return SubnetService(translations) class NetworkScanner: """Класс для сканирования подсети с использованием ARP-запросов. Назначение: Обнаруживает активные устройства в локальной сети, отправляя широковещательные ARP‑запросы по указанной подсети и собирая ответы (IP, MAC). Детали реализации: - Использует `scapy` (`Ether` + `ARP`) для генерации пакетов и `srp` для отправки/получения на канальном уровне от имени заданного интерфейса. - Таймаут ожидания ответа короткий (2 секунды), ориентирован на интерактивные сценарии UI. Ограничения и замечания: - Требуются права для отправки низкоуровневых пакетов (capabilities/root). - В сетях с фильтрацией широковещания количество обнаруженных устройств может быть ограничено. """ def __init__(self, interface: str, translations: Translations): """Инициализирует сканер сети. Устанавливает интерфейс для сканирования и объект переводов. Args: interface (str): Имя сетевого интерфейса (например, "wlp0s20f3"). translations (Translations): Объект переводов для текущего языка. """ self.interface = interface conf.iface = interface self.translations = translations def scan(self, subnet: str) -> List[Tuple[str, str]]: """Сканирует подсеть с использованием ARP-запросов. Отправляет ARP-пакеты для обнаружения активных устройств и возвращает список кортежей (IP, MAC). Args: subnet (str): Подсеть в формате CIDR (например, "10.1.1.0/24"). Returns: List[Tuple[str, str]]: Список кортежей (IP-адрес, MAC-адрес) обнаруженных устройств. Raises: Exception: Если произошла ошибка сканирования. """ try: arp = ARP(pdst=subnet) except Exception as e: logger.error(f"Error checking status: {str(e)}") try: arp = ARP(pdst=subnet) ether = Ether(dst="ff:ff:ff:ff:ff:ff") packet = ether / arp result = srp(packet, timeout=2, verbose=False, iface=self.interface)[0] return [(received.psrc, received.hwsrc) for sent, received in result] except Exception as e: logger.error(f"Error scanning subnet {subnet}: {str(e)}") return [] class PortScanner: """Класс для сканирования открытых портов на устройствах. Назначение: Плейсхолдер для будущего функционала активного сканирования TCP/UDP‑портов у обнаруженных узлов. В текущей версии возвращает пустой список. План развития: - Добавить неблокирующее сканирование наиболее распространённых портов (22, 80, 443 и т.д.) с ограничением по времени. - Настраиваемые профили глубины и скорости сканирования. """ def __init__(self, interface: str, translations: Translations): """Инициализирует сканер портов. Устанавливает интерфейс для сканирования и объект переводов. Args: interface (str): Имя сетевого интерфейса. translations (Translations): Объект переводов для текущего языка. """ self.interface = interface self.translations = translations def scan(self, ip: str) -> List[str]: """Сканирует открытые порты на указанном IP-адресе. Пока возвращает пустой список, так как функциональность не реализована. Args: ip (str): IP-адрес устройства для сканирования. Returns: List[str]: Список открытых портов (пока пустой). """ return [] class DeviceInfoCollector: """Класс для сбора информации об устройствах, включая вендоров и хостнеймы. Источники данных: - MacLookup: сопоставление префикса MAC (OUI) с названием производителя. - DNS: `gethostbyaddr` с малым таймаутом для обратного разрешения имени. - NetBIOS (nbtscan): опрос сегмента сети для извлечения NetBIOS‑имён. Политика кэширования: - Результаты `get_vendor` кэшируются с помощью `functools.lru_cache`. - Результаты `nbtscan` аккумулируются в словарях `nbtscan_cache`/`nbtscan_results` на время жизни экземпляра. """ def __init__(self, translations: Translations): """Инициализирует коллектор информации об устройствах. Настраивает кэширование и объект переводов. Args: translations (Translations): Объект переводов для текущего языка. """ self.mac_lookup = MacLookup() self.nbtscan_cache = {} self.nbtscan_results = {} self.translations = translations @lru_cache(maxsize=1000) def get_vendor(self, mac: str) -> str: """Получает вендора устройства по MAC-адресу. Использует кэширование для повышения производительности. Args: mac (str): MAC-адрес устройства. Returns: str: Название вендора или "Неизвестный" в случае ошибки. """ try: return self.mac_lookup.lookup(mac) except Exception as e: logger.warning(f"Failed to determine vendor for MAC {mac}: {str(e)}") return self.translations.gettext("Неизвестный") def scan_nbtscan(self, subnet: str) -> None: """Сканирует подсеть с помощью nbtscan для получения хостнеймов. Сохраняет результаты в кэш и логирует процесс. Args: subnet (str): Подсеть в формате CIDR. """ if subnet not in self.nbtscan_results: try: logger.debug(f"Starting nbtscan for subnet {subnet}") result = subprocess.run(["sudo", "nbtscan", "-r", str(subnet)], capture_output=True, text=True, check=False, timeout=10) logger.debug(f"Full nbtscan output: {result.stdout}") if result.returncode == 0: for line in result.stdout.splitlines(): logger.debug(f"Processing line: {line}") if not line.startswith(("IP address", "-", "Doing NBT")): parts = [p for p in line.split() if p] if len(parts) >= 2: ip_addr, hostname = parts[0], parts[1] if hostname != "": self.nbtscan_cache[ip_addr] = hostname logger.info(f"NetBIOS: Found hostname {hostname} for IP {ip_addr}") self.nbtscan_results[subnet] = True except subprocess.TimeoutExpired: logger.error(f"NetBIOS: Timeout for subnet {subnet}") except Exception as e: logger.error(f"NetBIOS: Error for subnet {subnet}: {str(e)}") def get_hostname(self, ip: str) -> Optional[str]: """Получает хостнейм для указанного IP-адреса. Пробует сначала DNS, затем NetBIOS. Args: ip (str): IP-адрес устройства. Returns: Optional[str]: Хостнейм или None, если не удалось определить. """ try: socket.setdefaulttimeout(2) return socket.gethostbyaddr(ip)[0] except (socket.herror, socket.gaierror, socket.timeout): logger.debug(f"DNS: Failed to resolve name for IP {ip}") subnet = ipaddress.ip_network(f"{ip}/24", strict=False) if subnet not in self.nbtscan_results: self.scan_nbtscan(str(subnet)) if ip in self.nbtscan_cache: return self.nbtscan_cache[ip] logger.debug(f"Failed to determine hostname for IP {ip} via DNS or NetBIOS") return None class DeviceStorage: """Класс для хранения и управления данными об устройствах. Формат файла подсети:: { "devices": [Device, ...] } Политика очистки: - Метод `_clean_old_devices` удаляет оффлайн‑устройства, не наблюдавшиеся более 7 дней, чтобы поддерживать актуальность списка. """ def __init__(self, base_path: str = "storage/subnets", translations: Translations = None): """Инициализирует хранилище устройств. Создаёт директорию, если её нет, и задаёт параметры очистки. Args: base_path (str): Путь к директории хранения данных. translations (Translations, optional): Объект переводов. """ self.base_path = Path(base_path) self.base_path.mkdir(parents=True, exist_ok=True) self.offline_timeout = timedelta(days=7) self.translations = translations or Translations.load("locales", ["ru"], "messages") # Fallback на русский def get_subnet_file(self, subnet: str) -> Path: """Возвращает путь к файлу для конкретной подсети. Заменяет символы '/' на '_' для корректного имени файла. Args: subnet (str): Подсеть в формате CIDR. Returns: Path: Путь к файлу. """ sanitized_subnet = subnet.replace("/", "_") return self.base_path / f"{sanitized_subnet}.json" def load_devices(self, subnet: str) -> Dict[str, Device]: """Загружает устройства для указанной подсети. Читает данные из JSON-файла и фильтрует устаревшие устройства. Args: subnet (str): Подсеть в формате CIDR. Returns: Dict[str, Device]: Словарь устройств с MAC как ключом. """ file_path = self.get_subnet_file(subnet) try: if file_path.exists(): with open(file_path, "r", encoding="utf-8") as f: data = json.load(f) if not isinstance(data, dict) or "devices" not in data: logger.warning(f"Invalid JSON structure in {file_path}, expected {'devices': [...]}, got {data}") return {} devices = {} for device_data in data.get("devices", []): device = Device(**device_data) if device.mac: devices[device.mac] = device return self._clean_old_devices(devices) return {} except json.JSONDecodeError as e: logger.error(f"JSON decode error in {file_path}: {str(e)}") return {} except Exception as e: logger.error(f"Error loading devices for subnet {subnet}: {str(e)}") return {} def _clean_old_devices(self, devices: Dict[str, Device]) -> Dict[str, Device]: """Очищает список устройств от устаревших оффлайн-устройств. Удаляет устройства, которые оффлайн более 7 дней. Args: devices (Dict[str, Device]): Словарь устройств. Returns: Dict[str, Device]: Отфильтрованный словарь устройств. """ now = datetime.now() filtered = {} for mac, device in devices.items(): if device.status == "offline" and device.last_seen: try: last_seen = datetime.fromisoformat(device.last_seen) if now - last_seen <= self.offline_timeout: filtered[mac] = device else: logger.info(f"Removed device {mac} (offline > 7 days)") except ValueError: logger.warning(f"Invalid last_seen format for device {mac}") else: filtered[mac] = device return filtered def save_devices(self, subnet: str, devices: List[Device]) -> None: """Сохраняет устройства для указанной подсети. Записывает данные в JSON-файл. Args: subnet (str): Подсеть в формате CIDR. devices (List[Device]): Список устройств для сохранения. Raises: Exception: Если произошла ошибка при сохранении. """ file_path = self.get_subnet_file(subnet) try: data = SubnetScanResult(devices=devices).model_dump() with open(file_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) logger.info(f"Saved {len(devices)} devices to {file_path}") except Exception as e: logger.error(f"Error saving devices for subnet {subnet}: {str(e)}") if self.translations: raise Exception(self.translations.gettext("Ошибка при сохранении устройств: {error}").format(error=str(e))) else: raise Exception(f"Ошибка при сохранении устройств: {str(e)}") class SubnetService: """Сервис для сканирования подсетей и управления устройствами. Роль сервиса: Объединяет сетевое сканирование (ARP), сбор дополнительной информации (вендор, хостнейм), слияние с кэшем предыдущих результатов и персистентное хранение данных по подсетям. Предоставляет вспомогательные операции для локального узла и работы с заметками устройств. Поток выполнения скана (`scan_subnet`): 1) Определение подсети (вручную или авто по интерфейсу). 2) Загрузка ранее сохранённых устройств подсети. 3) Добавление локального устройства (если принадлежит подсети). 4) ARP‑сканирование и параллельная обработка узлов (обогащение данных). 5) Маркировка оффлайн‑узлов из кэша, не встреченных в текущем скане. 6) Сортировка по IP и сохранение объединённого списка в JSON. """ def __init__(self, translations: Translations): """Инициализирует сервис подсети. Настраивает интерфейс, сканеры и хранилище с текущей локализацией. Args: translations (Translations): Объект переводов для текущего языка. """ self.interface = self._get_default_interface() self.network_scanner = NetworkScanner(self.interface, translations) self.port_scanner = PortScanner(self.interface, translations) self.device_info = DeviceInfoCollector(translations) self.storage = DeviceStorage(translations=translations) self.translations = translations logger.info(f"Initialized SubnetService with interface: {self.interface}") def _get_default_interface(self) -> str: """Определяет интерфейс по умолчанию. Использует информацию о шлюзе или первый доступный интерфейс. Returns: str: Имя интерфейса (например, "wlp0s20f3" или "eth0"). Raises: Exception: Если не удалось определить интерфейс. """ try: gateways = netifaces.gateways() default_gateway = gateways.get("default", {}).get(netifaces.AF_INET) if default_gateway: iface = default_gateway[1] logger.info(f"Default interface from gateway: {iface}") return iface interfaces = netifaces.interfaces() for iface in interfaces: if iface != "lo": addrs = netifaces.ifaddresses(iface) if netifaces.AF_INET in addrs: logger.info(f"Selected interface: {iface}") return iface logger.warning("No suitable interface found, defaulting to eth0") return "eth0" except Exception as e: logger.warning(f"Failed to determine interface: {str(e)}, defaulting to eth0") return "eth0" def _get_default_subnet(self) -> str: """Автоматически определяет подсеть локального устройства. Использует информацию о текущем интерфейсе для определения подсети. Returns: str: Подсеть в формате CIDR (например, "10.1.1.0/24"). Raises: Exception: Если не удалось определить подсеть. """ try: interfaces = netifaces.interfaces() for iface in interfaces: if iface == "lo" or iface != self.interface: continue addrs = netifaces.ifaddresses(iface) if netifaces.AF_INET in addrs: ip_info = addrs[netifaces.AF_INET][0] ip = ip_info["addr"] netmask = ip_info["netmask"] # Преобразуем маску из строки (например, "255.255.255.0") в CIDR mask = ipaddress.ip_network(f"0.0.0.0/{netmask}", strict=False).prefixlen network = ipaddress.ip_network(f"{ip}/{mask}", strict=False) logger.info(f"Auto-detected subnet: {network} (IP: {ip}, Mask: {netmask} -> /{mask})") return str(network) logger.warning("No suitable subnet found, defaulting to 0.0.0.0/0") return "0.0.0.0/0" except Exception as e: logger.error(f"Error determining default subnet: {str(e)}") return "0.0.0.0/0" def scan_subnet(self, subnet: Optional[str] = None) -> SubnetScanResult: """Сканирует указанную подсеть и возвращает результаты. Выполняет ARP-сканирование, собирает информацию о устройствах и сохраняет данные. Args: subnet (Optional[str]): Подсеть в формате CIDR (например, "10.1.1.0/24"). Если None или пустая строка, определяется автоматически. Returns: SubnetScanResult: Объект с результатами сканирования, содержащий список устройств или сообщение об ошибке. Raises: Exception: Если произошла ошибка сканирования. """ start_time = time.time() devices = [] now = datetime.now().astimezone() formatted_now = now.strftime("%d-%m-%Y, %H:%M") try: # Определяем подсеть, если не указана или пустая if not subnet or subnet.strip() == "": subnet = self._get_default_subnet() if subnet == "0.0.0.0/0": logger.warning("Default subnet is invalid, scan aborted") return SubnetScanResult(devices=devices, error=self.translations.gettext("Недопустимая подсеть по умолчанию")) saved_devices = self.storage.load_devices(subnet) local_device = self._get_local_device(subnet) if local_device: local_device.status = "online" local_device.last_seen = formatted_now devices.append(local_device) if local_device.mac: saved_devices[local_device.mac] = local_device logger.info(f"Detected local device: {local_device.ip}, MAC: {local_device.mac}") logger.info(f"Scanning subnet: {subnet} with interface: {self.interface}") discovered_devices = self.network_scanner.scan(subnet) logger.info(f"Discovered devices: {len(discovered_devices)}") self.device_info.scan_nbtscan(subnet) if not discovered_devices and not local_device: logger.warning(f"No devices found in subnet {subnet}") return SubnetScanResult(devices=[], error=self.translations.gettext("В данной подсети не обнаружено устройств")) def process_device(ip_mac: Tuple[str, str]) -> Optional[Device]: ip, mac = ip_mac if local_device and ip == local_device.ip: return None existing_device = saved_devices.get(mac) if existing_device: return Device( ip=ip, mac=mac, hostname=self.device_info.get_hostname(ip) or existing_device.hostname, vendor=self.device_info.get_vendor(mac), open_ports=existing_device.open_ports, status="online", last_seen=formatted_now, note=existing_device.note, ) vendor = self.device_info.get_vendor(mac) hostname = self.device_info.get_hostname(ip) return Device( ip=ip, mac=mac, hostname=hostname, vendor=vendor, open_ports=[], status="online", last_seen=formatted_now, note=None, ) with ThreadPoolExecutor(max_workers=10) as executor: device_results = executor.map(process_device, discovered_devices) devices.extend([d for d in device_results if d is not None]) active_macs = {d.mac for d in devices if d.mac} for mac, device in saved_devices.items(): if mac not in active_macs: device.status = "offline" devices.append(device) devices.sort(key=lambda x: int(ipaddress.IPv4Address(x.ip))) for device in devices: if device.mac: saved_devices[device.mac] = device self.storage.save_devices(subnet, list(saved_devices.values())) logger.info( f"Subnet scan {subnet} completed in {time.time() - start_time:.2f} seconds, " f"found {len(devices)} devices (active: {len(active_macs)})" ) return SubnetScanResult(devices=devices) except Exception as e: logger.error(f"Error scanning subnet {subnet}: {str(e)}") return SubnetScanResult(devices=devices, error=self.translations.gettext("Ошибка сканирования: {error}").format(error=str(e))) def _get_local_device(self, subnet: str) -> Optional[Device]: """Определяет локальное устройство в указанной подсети. Использует информацию об интерфейсе для обнаружения. Args: subnet (str): Подсеть в формате CIDR. Returns: Optional[Device]: Объект устройства или None, если не найдено. """ now = datetime.now().astimezone() formatted_now = now.strftime("%d-%m-%Y, %H:%M") try: subnet_network = ipaddress.ip_network(subnet, strict=False) interfaces = netifaces.interfaces() for iface in interfaces: if iface == "lo": continue addrs = netifaces.ifaddresses(iface) if netifaces.AF_INET in addrs and netifaces.AF_LINK in addrs: ip = addrs[netifaces.AF_INET][0]["addr"] if ipaddress.ip_address(ip) in subnet_network: mac = addrs[netifaces.AF_LINK][0]["addr"] vendor = self.device_info.get_vendor(mac) hostname = self.device_info.get_hostname(ip) logger.info(f"Local device detected: IP={ip}, MAC={mac}, Interface={iface}") return Device( ip=ip, mac=mac, hostname=hostname, vendor=vendor, open_ports=[], status="online", last_seen=formatted_now, note=None, ) logger.warning(f"Local device not found in subnet {subnet}") return None except Exception as e: logger.warning(f"Failed to get local device: {str(e)}") return None def add_note(self, mac: str, note: str, translations: Translations) -> Tuple[bool, Optional[Tuple[str, Dict]]]: """Добавляет заметку для устройства по MAC-адресу. Привязывает заметку к устройству, если оно существует. Args: mac (str): MAC-адрес устройства. note (str): Текст заметки. translations (Translations): Объект переводов для текущего языка. Returns: Tuple[bool, Optional[Tuple[str, Dict]]]: Кортеж (успех, результат), где результат содержит код ошибки/успеха и данные. """ try: saved_devices = {} for subnet in self.storage.base_path.glob("*.json"): devices = self.storage.load_devices(subnet.stem.replace("_", "/")) saved_devices.update(devices) if mac not in saved_devices: logger.warning(f"Device with MAC {mac} not found") return False, ("not_found", {"mac": mac}) if saved_devices[mac].note is not None: logger.error(f"Note for device with MAC {mac} already exists") return False, ("note_exists", {"mac": mac}) for subnet_file in self.storage.base_path.glob("*.json"): subnet = subnet_file.stem.replace("_", "/") devices = self.storage.load_devices(subnet) if mac in devices: devices[mac].note = note self.storage.save_devices(subnet, list(devices.values())) break logger.info(f"Added note '{note}' for device with MAC {mac}") return True, ("success", {"mac": mac, "note": note}) except Exception as e: logger.error(f"Error adding note for MAC {mac}: {str(e)}") return False, ("internal_error", {"error": str(e)}) def update_note(self, mac: str, note: str, translations: Translations) -> Tuple[bool, Optional[Tuple[str, Dict]]]: """Обновляет заметку для устройства по MAC-адресу. Изменяет существующую заметку, если она есть. Args: mac (str): MAC-адрес устройства. note (str): Новый текст заметки. translations (Translations): Объект переводов для текущего языка. Returns: Tuple[bool, Optional[Tuple[str, Dict]]]: Кортеж (успех, результат), где результат содержит код ошибки/успеха и данные. """ try: saved_devices = {} for subnet in self.storage.base_path.glob("*.json"): devices = self.storage.load_devices(subnet.stem.replace("_", "/")) saved_devices.update(devices) if mac not in saved_devices: logger.warning(f"Device with MAC {mac} not found") return False, ("not_found", {"mac": mac}) if saved_devices[mac].note is None: logger.error(f"Note for device with MAC {mac} does not exist") return False, ("note_not_exists", {"mac": mac}) for subnet_file in self.storage.base_path.glob("*.json"): subnet = subnet_file.stem.replace("_", "/") devices = self.storage.load_devices(subnet) if mac in devices: devices[mac].note = note self.storage.save_devices(subnet, list(devices.values())) break logger.info(f"Updated note '{note}' for device with MAC {mac}") return True, ("success", {"mac": mac, "note": note}) except Exception as e: logger.error(f"Error updating note for MAC {mac}: {str(e)}") return False, ("internal_error", {"error": str(e)}) def delete_note(self, mac: str, translations: Translations) -> Tuple[bool, Optional[Tuple[str, Dict]]]: """Удаляет заметку для устройства по MAC-адресу. Удаляет привязанную заметку, если она существует. Args: mac (str): MAC-адрес устройства. translations (Translations): Объект переводов для текущего языка. Returns: Tuple[bool, Optional[Tuple[str, Dict]]]: Кортеж (успех, результат), где результат содержит код ошибки/успеха и данные. """ try: saved_devices = {} for subnet in self.storage.base_path.glob("*.json"): devices = self.storage.load_devices(subnet.stem.replace("_", "/")) saved_devices.update(devices) if mac not in saved_devices: logger.warning(f"Device with MAC {mac} not found") return False, ("not_found", {"mac": mac}) if saved_devices[mac].note is None: logger.error(f"Note for device with MAC {mac} does not exist") return False, ("note_not_exists", {"mac": mac}) for subnet_file in self.storage.base_path.glob("*.json"): subnet = subnet_file.stem.replace("_", "/") devices = self.storage.load_devices(subnet) if mac in devices: devices[mac].note = None self.storage.save_devices(subnet, list(devices.values())) break logger.info(f"Deleted note for device with MAC {mac}") return True, ("success", {"mac": mac}) except Exception as e: logger.error(f"Error deleting note for MAC {mac}: {str(e)}") return False, ("internal_error", {"error": str(e)})