594 lines
No EOL
27 KiB
Python
594 lines
No EOL
27 KiB
Python
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 models.subnet import SubnetScanResult, Device
|
||
from config.ports import port_config
|
||
from scapy.layers.inet import IP, TCP
|
||
from scapy.layers.l2 import ARP, Ether
|
||
from scapy.sendrecv import srp, sr1
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class NetworkScanner:
|
||
"""Класс для сканирования подсети с использованием ARP-запросов.
|
||
|
||
Выполняет обнаружение устройств в указанной подсети через ARP-пакеты.
|
||
"""
|
||
|
||
def __init__(self, interface: str):
|
||
"""Инициализирует сканер сети с указанным сетевым интерфейсом.
|
||
|
||
Args:
|
||
interface (str): Название сетевого интерфейса (например, "eth0").
|
||
"""
|
||
self.interface = interface
|
||
conf.iface = interface
|
||
|
||
def scan(self, subnet: str) -> List[Tuple[str, str]]:
|
||
"""Сканирует подсеть с использованием ARP-запросов.
|
||
|
||
Args:
|
||
subnet (str): Подсеть в формате CIDR (например, "192.168.1.0/24").
|
||
|
||
Returns:
|
||
List[Tuple[str, str]]: Список кортежей (IP-адрес, MAC-адрес) обнаруженных
|
||
устройств.
|
||
|
||
Raises:
|
||
Exception: Если произошла ошибка при отправке ARP-пакетов.
|
||
"""
|
||
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"Ошибка ARP-сканирования подсети {subnet}: {str(e)}")
|
||
return []
|
||
|
||
class PortScanner:
|
||
"""Класс для сканирования открытых портов на устройствах.
|
||
|
||
Выполняет сканирование портов с использованием TCP SYN-запросов.
|
||
"""
|
||
|
||
def __init__(self, interface: str):
|
||
"""Инициализирует сканер портов с указанным сетевым интерфейсом.
|
||
|
||
Args:
|
||
interface (str): Название сетевого интерфейса (например, "eth0").
|
||
"""
|
||
self.interface = interface
|
||
|
||
def scan(self, ip: str) -> List[str]:
|
||
"""Сканирует открытые порты на указанном IP-адресе.
|
||
|
||
Args:
|
||
ip (str): IP-адрес устройства для сканирования.
|
||
|
||
Returns:
|
||
List[str]: Список имён открытых портов (например, "HTTP", "SSH").
|
||
|
||
Raises:
|
||
Exception: Если произошла ошибка при сканировании портов.
|
||
"""
|
||
open_ports = []
|
||
|
||
def check_port(port: int) -> Optional[str]:
|
||
"""Проверяет доступность конкретного порта.
|
||
|
||
Args:
|
||
port (int): Номер порта для проверки.
|
||
|
||
Returns:
|
||
Optional[str]: Имя порта, если открыт, иначе None.
|
||
"""
|
||
try:
|
||
pkt = sr1(
|
||
IP(dst=ip) / TCP(dport=port, flags="S"),
|
||
timeout=2,
|
||
verbose=False,
|
||
)
|
||
if pkt and pkt.haslayer(TCP) and pkt[TCP].flags == 0x12:
|
||
return port_config.get_port_name(port)
|
||
return None
|
||
except Exception as e:
|
||
logger.debug(
|
||
f"Ошибка проверки порта {port} для IP {ip}: {str(e)}")
|
||
return None
|
||
|
||
try:
|
||
max_workers = min(len(port_config.get_ports()), 15)
|
||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||
results = executor.map(check_port, port_config.get_ports())
|
||
open_ports = [port for port in results if port is not None]
|
||
logger.debug(f"Открытые порты для IP {ip}: {open_ports}")
|
||
return open_ports
|
||
except Exception as e:
|
||
logger.warning(
|
||
f"Ошибка при сканировании портов для IP {ip}: {str(e)}")
|
||
return []
|
||
|
||
class DeviceInfoCollector:
|
||
"""Класс для сбора информации об устройствах, включая вендоров и хостнеймы.
|
||
|
||
Управляет кэшированием данных NetBIOS и DNS.
|
||
"""
|
||
|
||
def __init__(self):
|
||
"""Инициализирует коллектор информации об устройствах.
|
||
|
||
Создаёт объект MacLookup для определения вендоров и инициализирует кэши.
|
||
"""
|
||
self.mac_lookup = MacLookup()
|
||
self.nbtscan_cache = {}
|
||
self.nbtscan_results = {}
|
||
|
||
@lru_cache(maxsize=1000)
|
||
def get_vendor(self, mac: str) -> str:
|
||
"""Определяет вендора по MAC-адресу с использованием кэширования.
|
||
|
||
Args:
|
||
mac (str): MAC-адрес устройства в формате XX:XX:XX:XX:XX:XX.
|
||
|
||
Returns:
|
||
str: Название вендора или "Unknown", если вендор не определён.
|
||
|
||
Raises:
|
||
Exception: Если произошла ошибка при запросе к базе вендоров.
|
||
"""
|
||
try:
|
||
return self.mac_lookup.lookup(mac)
|
||
except Exception as e:
|
||
logger.warning(
|
||
f"Не удалось определить производителя для MAC {mac}: {str(e)}")
|
||
return "Unknown"
|
||
|
||
def scan_nbtscan(self, subnet: str) -> None:
|
||
"""Выполняет сканирование NetBIOS для указанной подсети.
|
||
|
||
Args:
|
||
subnet (str): Подсеть в формате CIDR (например, "192.168.1.0/24").
|
||
|
||
Сохраняет результаты в кэш, игнорируя хостнеймы "<unknown>".
|
||
"""
|
||
if subnet not in self.nbtscan_results:
|
||
try:
|
||
logger.debug(f"Запуск nbtscan для подсети {subnet}")
|
||
result = subprocess.run(["sudo", "nbtscan", "-r", str(subnet)],
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
timeout=10)
|
||
logger.debug(f"Полный вывод nbtscan: {result.stdout}")
|
||
if result.returncode == 0:
|
||
for line in result.stdout.splitlines():
|
||
logger.debug(f"Обработка строки: {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 != "<unknown>":
|
||
self.nbtscan_cache[ip_addr] = hostname
|
||
logger.info(f"NetBIOS: Найден хостнейм {hostname} для IP {ip_addr}")
|
||
self.nbtscan_results[subnet] = True
|
||
except subprocess.TimeoutExpired:
|
||
logger.error(f"NetBIOS: Таймаут для подсети {subnet}")
|
||
except Exception as e:
|
||
logger.error(f"NetBIOS: Ошибка для подсети {subnet}: {str(e)}")
|
||
|
||
def get_hostname(self, ip: str) -> Optional[str]:
|
||
"""Определяет хостнейм для указанного IP-адреса.
|
||
|
||
Args:
|
||
ip (str): IP-адрес устройства.
|
||
|
||
Returns:
|
||
Optional[str]: Хостнейм, если найден, иначе None.
|
||
|
||
Сначала пытается использовать DNS, затем NetBIOS через nbtscan.
|
||
"""
|
||
try:
|
||
socket.setdefaulttimeout(2)
|
||
return socket.gethostbyaddr(ip)[0]
|
||
except (socket.herror, socket.gaierror, socket.timeout):
|
||
logger.debug(f"DNS: Не удалось разрешить имя для 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"Не удалось определить хостнейм для IP {ip} через DNS или NetBIOS")
|
||
return None
|
||
|
||
class DeviceStorage:
|
||
"""Класс для хранения и управления данными об устройствах.
|
||
|
||
Хранит устройства в JSON-файле и удаляет устаревшие записи.
|
||
"""
|
||
|
||
def __init__(self, storage_path: str = "storage/devices/devices.json"):
|
||
"""Инициализирует хранилище устройств с указанным путём.
|
||
|
||
Args:
|
||
storage_path (str, optional): Путь к файлу хранения.
|
||
По умолчанию "storage/devices/devices.json".
|
||
|
||
Создаёт родительские директории, если они не существуют.
|
||
"""
|
||
self.storage_file = Path(storage_path)
|
||
self.storage_file.parent.mkdir(parents=True, exist_ok=True)
|
||
self.offline_timeout = timedelta(days=7)
|
||
|
||
def load_devices(self) -> Dict[str, Device]:
|
||
"""Загружает устройства из файла хранения.
|
||
|
||
Returns:
|
||
Dict[str, Device]: Словарь устройств, где ключ — MAC-адрес.
|
||
|
||
Удаляет устройства с неверным форматом last_seen и возвращает пустой словарь
|
||
при ошибке.
|
||
"""
|
||
try:
|
||
if self.storage_file.exists():
|
||
with open(self.storage_file, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
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 Exception as e:
|
||
logger.error(f"Ошибка при загрузке устройств: {str(e)}")
|
||
return {}
|
||
|
||
def _clean_old_devices(self, devices: Dict[str, Device]) -> Dict[str, Device]:
|
||
"""Очищает устаревшие устройства из словаря.
|
||
|
||
Args:
|
||
devices (Dict[str, Device]): Словарь устройств с MAC-адресами как ключами.
|
||
|
||
Returns:
|
||
Dict[str, Device]: Отфильтрованный словарь, исключающий устройства
|
||
offline более 7 дней.
|
||
"""
|
||
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"Удалено устройство {mac} (offline > 7 дней)")
|
||
except ValueError:
|
||
logger.warning(f"Неверный формат last_seen для устройства {mac}")
|
||
else:
|
||
filtered[mac] = device
|
||
return filtered
|
||
|
||
def save_devices(self, devices: List[Device]) -> None:
|
||
"""Сохраняет список устройств в файл.
|
||
|
||
Args:
|
||
devices (List[Device]): Список объектов Device для сохранения.
|
||
|
||
Raises:
|
||
Exception: Если произошла ошибка при записи в файл.
|
||
"""
|
||
try:
|
||
data = SubnetScanResult(devices=devices).model_dump()
|
||
with open(self.storage_file, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||
logger.info(f"Сохранено {len(devices)} устройств в {self.storage_file}")
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при сохранении устройств: {str(e)}")
|
||
|
||
class SubnetService:
|
||
"""Сервис для сканирования подсетей и управления устройствами.
|
||
|
||
Объединяет функционал сканирования сети, портов и хранения данных об устройствах.
|
||
"""
|
||
|
||
def __init__(self):
|
||
"""Инициализирует сервис подсети с настройкой всех компонентов.
|
||
|
||
Автоматически определяет интерфейс и инициализирует сканеры и хранилище.
|
||
"""
|
||
self.interface = self._get_default_interface()
|
||
self.network_scanner = NetworkScanner(self.interface)
|
||
self.port_scanner = PortScanner(self.interface)
|
||
self.device_info = DeviceInfoCollector()
|
||
self.storage = DeviceStorage()
|
||
|
||
def _get_default_interface(self) -> str:
|
||
"""Определяет имя сетевого интерфейса по умолчанию.
|
||
|
||
Returns:
|
||
str: Название интерфейса (например, "eth0").
|
||
|
||
Использует информацию о шлюзе или первый доступный интерфейс, отличный от "lo".
|
||
"""
|
||
try:
|
||
gateways = netifaces.gateways()
|
||
default_gateway = gateways.get("default", {}).get(netifaces.AF_INET)
|
||
if default_gateway:
|
||
return default_gateway[1]
|
||
interfaces = netifaces.interfaces()
|
||
for iface in interfaces:
|
||
if iface != "lo":
|
||
return iface
|
||
return "eth0"
|
||
except Exception as e:
|
||
logger.warning(f"Не удалось определить интерфейс: {str(e)}")
|
||
return "eth0"
|
||
|
||
def scan_subnet(self, subnet: str, ports: List[int] = None) -> SubnetScanResult:
|
||
"""Сканирует указанную подсеть и возвращает результаты.
|
||
|
||
Args:
|
||
subnet (str): Подсеть в формате CIDR (например, "192.168.1.0/24").
|
||
ports (List[int], optional): Список портов для сканирования.
|
||
По умолчанию None (используются порты из config.ports).
|
||
|
||
Returns:
|
||
SubnetScanResult: Объект с результатами сканирования, содержащий список
|
||
устройств.
|
||
|
||
Сохраняет данные в хранилище и обновляет статус устройств.
|
||
"""
|
||
start_time = time.time()
|
||
devices = []
|
||
now = datetime.now().astimezone()
|
||
formatted_now = now.strftime("%d-%m-%Y, %H:%M")
|
||
try:
|
||
saved_devices = self.storage.load_devices()
|
||
|
||
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
|
||
|
||
discovered_devices = self.network_scanner.scan(subnet)
|
||
self.device_info.scan_nbtscan(subnet)
|
||
|
||
def process_device(ip_mac: Tuple[str, str]) -> Optional[Device]:
|
||
"""Обрабатывает устройство из пары IP и MAC.
|
||
|
||
Args:
|
||
ip_mac (Tuple[str, str]): Кортеж с IP-адресом и MAC-адресом.
|
||
|
||
Returns:
|
||
Optional[Device]: Объект Device или None, если устройство локальное.
|
||
"""
|
||
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=self.port_scanner.scan(ip) or 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)
|
||
open_ports = self.port_scanner.scan(ip)
|
||
return Device(
|
||
ip=ip,
|
||
mac=mac,
|
||
hostname=hostname,
|
||
vendor=vendor,
|
||
open_ports=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(list(saved_devices.values()))
|
||
|
||
logger.info(
|
||
f"Сканирование подсети {subnet} завершено за {time.time() - start_time:.2f} секунд, "
|
||
f"найдено {len(devices)} устройств (активных: {len(active_macs)})"
|
||
)
|
||
return SubnetScanResult(devices=devices)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при сканировании подсети {subnet}: {str(e)}")
|
||
return SubnetScanResult(devices=devices)
|
||
|
||
def _get_local_device(self, subnet: str) -> Optional[Device]:
|
||
"""Определяет локальное устройство в указанной подсети.
|
||
|
||
Args:
|
||
subnet (str): Подсеть в формате CIDR (например, "192.168.1.0/24").
|
||
|
||
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)
|
||
open_ports = self.port_scanner.scan(ip)
|
||
logger.debug(
|
||
f"Локальное устройство: IP={ip}, MAC={mac}, Ports={open_ports}, Hostname={hostname}"
|
||
)
|
||
return Device(
|
||
ip=ip,
|
||
mac=mac,
|
||
hostname=hostname,
|
||
vendor=vendor,
|
||
open_ports=open_ports,
|
||
status="online",
|
||
last_seen=formatted_now,
|
||
note=None,
|
||
)
|
||
logger.warning(
|
||
f"Локальное устройство не найдено в подсети {subnet}")
|
||
return None
|
||
except Exception as e:
|
||
logger.warning(
|
||
f"Не удалось получить локальное устройство: {str(e)}")
|
||
return None
|
||
|
||
def add_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]:
|
||
"""Добавляет заметку к устройству по MAC-адресу.
|
||
|
||
Args:
|
||
mac (str): MAC-адрес устройства.
|
||
note (str): Текст заметки.
|
||
|
||
Returns:
|
||
Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке).
|
||
- Успех: (True, None).
|
||
- Ошибка: (False, сообщение об ошибке).
|
||
|
||
Raises:
|
||
Exception: Если произошла ошибка при сохранении.
|
||
"""
|
||
try:
|
||
saved_devices = self.storage.load_devices()
|
||
if mac not in saved_devices:
|
||
logger.warning(f"Устройство с MAC {mac} не найдено")
|
||
return False, f"Устройство с MAC {mac} не найдено"
|
||
if saved_devices[mac].note is not None:
|
||
logger.error(
|
||
f"Заметка для устройства с MAC {mac} уже существует")
|
||
return False, f"Заметка для устройства с MAC {mac} уже существует"
|
||
saved_devices[mac].note = note
|
||
self.storage.save_devices(list(saved_devices.values()))
|
||
logger.info(
|
||
f"Добавлена заметка '{note}' для устройства с MAC {mac}")
|
||
return True, None
|
||
except Exception as e:
|
||
logger.error(
|
||
f"Ошибка при добавлении заметки для MAC {mac}: {str(e)}")
|
||
return False, f"Внутренняя ошибка сервера: {str(e)}"
|
||
|
||
def update_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]:
|
||
"""Обновляет заметку устройства по MAC-адресу.
|
||
|
||
Args:
|
||
mac (str): MAC-адрес устройства.
|
||
note (str): Новый текст заметки.
|
||
|
||
Returns:
|
||
Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке).
|
||
- Успех: (True, None).
|
||
- Ошибка: (False, сообщение об ошибке).
|
||
|
||
Raises:
|
||
Exception: Если произошла ошибка при сохранении.
|
||
"""
|
||
try:
|
||
saved_devices = self.storage.load_devices()
|
||
if mac not in saved_devices:
|
||
logger.warning(f"Устройство с MAC {mac} не найдено")
|
||
return False, f"Устройство с MAC {mac} не найдено"
|
||
if saved_devices[mac].note is None:
|
||
logger.error(
|
||
f"Заметка для устройства с MAC {mac} не существует")
|
||
return False, f"Заметка для устройства с MAC {mac} не существует"
|
||
saved_devices[mac].note = note
|
||
self.storage.save_devices(list(saved_devices.values()))
|
||
logger.info(
|
||
f"Обновлена заметка '{note}' для устройства с MAC {mac}")
|
||
return True, None
|
||
except Exception as e:
|
||
logger.error(
|
||
f"Ошибка при обновлении заметки для MAC {mac}: {str(e)}")
|
||
return False, f"Внутренняя ошибка сервера: {str(e)}"
|
||
|
||
def delete_note(self, mac: str) -> Tuple[bool, Optional[str]]:
|
||
"""Удаляет заметку устройства по MAC-адресу.
|
||
|
||
Args:
|
||
mac (str): MAC-адрес устройства.
|
||
|
||
Returns:
|
||
Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке).
|
||
- Успех: (True, None).
|
||
- Ошибка: (False, сообщение об ошибке).
|
||
|
||
Raises:
|
||
Exception: Если произошла ошибка при сохранении.
|
||
"""
|
||
try:
|
||
saved_devices = self.storage.load_devices()
|
||
if mac not in saved_devices:
|
||
logger.warning(f"Устройство с MAC {mac} не найдено")
|
||
return False, f"Устройство с MAC {mac} не найдено"
|
||
if saved_devices[mac].note is None:
|
||
logger.error(
|
||
f"Заметка для устройства с MAC {mac} не существует")
|
||
return False, f"Заметка для устройства с MAC {mac} не существует"
|
||
saved_devices[mac].note = None
|
||
self.storage.save_devices(list(saved_devices.values()))
|
||
logger.info(f"Удалена заметка для устройства с MAC {mac}")
|
||
return True, None
|
||
except Exception as e:
|
||
logger.error(
|
||
f"Ошибка при удалении заметки для MAC {mac}: {str(e)}")
|
||
return False, f"Внутренняя ошибка сервера: {str(e)}" |