HybrydMain/gonets
3
2
Fork 0
Code Issues 5 Pull requests Projects 1 Releases 3 Packages Wiki Activity Actions

Последняя версия сканера устройств (хранение истории, вывод портов и вывод последнего входа)

Browse source
This commit is contained in:
Галанский Сергей Александрович 2025-06-09 11:36:57 +05:00
parent d67ea17fd7
commit 6b6239ef9d
1 changed files with 55 additions and 82 deletions
Download patch file Download diff file Expand all files Collapse all files

137
SRV/services/subnet_service.py
View file

@ -1,5 +1,5 @@
from scapy.config import conf from scapy.config import conf
from manuf import MacParser from mac_vendor_lookup import MacLookup
from typing import List, Optional, Tuple, Dict from typing import List, Optional, Tuple, Dict
import socket import socket
import logging import logging
@ -12,7 +12,6 @@ import json
from pathlib import Path from pathlib import Path
from datetime import datetime, timedelta from datetime import datetime, timedelta
from zoneinfo import ZoneInfo from zoneinfo import ZoneInfo
import os
from models.subnet import SubnetScanResult, Device from models.subnet import SubnetScanResult, Device
from config.ports import port_config from config.ports import port_config
@ -23,17 +22,6 @@ from scapy.sendrecv import srp, sr1
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Подавляем логи Scapy
logging.getLogger("scapy").setLevel(logging.ERROR)
def normalize_mac(mac: str) -> str:
"""Нормализует MAC-адрес к формату XX:XX:XX:XX:XX:XX."""
if not mac:
return ""
mac = mac.replace(":", "").replace("-", "").upper()
return ":".join(mac[i:i + 2] for i in range(0, 12, 2))
class NetworkScanner: class NetworkScanner:
"""Класс для сканирования сети с помощью ARP-запросов.""" """Класс для сканирования сети с помощью ARP-запросов."""
@ -41,19 +29,14 @@ class NetworkScanner:
def __init__(self, interface: str): def __init__(self, interface: str):
self.interface = interface self.interface = interface
conf.iface = interface conf.iface = interface
self.arp_cache: Dict[str, str] = {}
def scan(self, subnet: str) -> List[Tuple[str, str]]: def scan(self, subnet: str) -> List[Tuple[str, str]]:
try: try:
arp = ARP(pdst=subnet) arp = ARP(pdst=subnet)
ether = Ether(dst="ff:ff:ff:ff:ff:ff") ether = Ether(dst="ff:ff:ff:ff:ff:ff")
packet = ether / arp packet = ether / arp
result = srp(packet, timeout=5, verbose=False, iface=self.interface)[0] result = srp(packet, timeout=2, verbose=False, iface=self.interface)[0]
devices = [(received.psrc, normalize_mac(received.hwsrc)) for sent, received in result] return [(received.psrc, received.hwsrc) for sent, received in result]
for ip, mac in devices:
self.arp_cache[ip] = mac
logger.debug(f"ARP-кэш обновлён: {len(self.arp_cache)} записей")
return devices
except Exception as e: except Exception as e:
logger.error(f"Ошибка ARP-сканирования подсети {subnet}: {str(e)}") logger.error(f"Ошибка ARP-сканирования подсети {subnet}: {str(e)}")
return [] return []
@ -62,18 +45,16 @@ class NetworkScanner:
class PortScanner: class PortScanner:
"""Класс для сканирования портов на устройстве.""" """Класс для сканирования портов на устройстве."""
def __init__(self, interface: str, arp_cache: Dict[str, str]): def __init__(self, interface: str):
self.interface = interface self.interface = interface
self.arp_cache = arp_cache
def scan(self, ip: str) -> List[str]: def scan(self, ip: str) -> List[str]:
open_ports = [] open_ports = []
def check_port(port: int) -> Optional[str]: def check_port(port: int) -> Optional[str]:
try: try:
dst_mac = self.arp_cache.get(ip, "ff:ff:ff:ff:ff:ff")
pkt = sr1( pkt = sr1(
Ether(dst=dst_mac) / IP(dst=ip) / TCP(dport=port, flags="S"), IP(dst=ip) / TCP(dport=port, flags="S"),
timeout=2, timeout=2,
verbose=False, verbose=False,
iface=self.interface iface=self.interface
@ -86,7 +67,7 @@ class PortScanner:
return None return None
try: try:
max_workers = min(len(port_config.get_ports()), 20) # Уменьшено до 20 max_workers = min(len(port_config.get_ports()), 15) # Снижено с 100
with ThreadPoolExecutor(max_workers=max_workers) as executor: with ThreadPoolExecutor(max_workers=max_workers) as executor:
results = executor.map(check_port, port_config.get_ports()) results = executor.map(check_port, port_config.get_ports())
open_ports = [port for port in results if port is not None] open_ports = [port for port in results if port is not None]
@ -101,33 +82,14 @@ class DeviceInfoCollector:
"""Класс для сбора информации об устройстве.""" """Класс для сбора информации об устройстве."""
def __init__(self): def __init__(self):
self.mac_lookup = MacParser() self.mac_lookup = MacLookup()
try:
logger.info("Обновление базы производителей MAC...")
self.mac_lookup.update()
logger.info("База производителей обновлена")
except Exception as e:
logger.warning(f"Не удалось обновить базу производителей: {str(e)}")
# Проверяем наличие локальной базы
manuf_db = Path(os.path.expanduser("~/.cache/manuf/manuf.txt"))
if manuf_db.exists():
logger.info(f"Используется локальная база OUI: {manuf_db}")
else:
logger.error("Локальная база OUI отсутствует, vendors могут быть Unknown")
@lru_cache(maxsize=1000) @lru_cache(maxsize=1000) # Снижено с 10000
def get_vendor(self, mac: str) -> str: def get_vendor(self, mac: str) -> str:
if not mac or mac.lower() == "ff:ff:ff:ff:ff:ff":
return "Unknown"
normalized_mac = normalize_mac(mac)
try: try:
vendor = self.mac_lookup.get_manuf(normalized_mac) return self.mac_lookup.lookup(mac)
if vendor is None:
vendor = "Unknown"
logger.debug(f"Vendor для MAC {normalized_mac}: {vendor}")
return vendor
except Exception as e: except Exception as e:
logger.debug(f"Не удалось определить производителя для MAC {normalized_mac}: {str(e)}") logger.warning(f"Не удалось определить производителя для MAC {mac}: {str(e)}")
return "Unknown" return "Unknown"
def get_hostname(self, ip: str) -> Optional[str]: def get_hostname(self, ip: str) -> Optional[str]:
@ -154,7 +116,7 @@ class DeviceStorage:
for device_data in data.get("devices", []): for device_data in data.get("devices", []):
device = Device(**device_data) device = Device(**device_data)
if device.mac: if device.mac:
devices[normalize_mac(device.mac)] = device devices[device.mac] = device
return self._clean_old_devices(devices) return self._clean_old_devices(devices)
return {} return {}
except Exception as e: except Exception as e:
@ -180,11 +142,6 @@ class DeviceStorage:
def save_devices(self, devices: List[Device]) -> None: def save_devices(self, devices: List[Device]) -> None:
try: try:
for device in devices:
if isinstance(device.last_seen, datetime):
device.last_seen = device.last_seen.isoformat()
if device.mac:
device.mac = normalize_mac(device.mac)
data = SubnetScanResult(devices=devices).model_dump() data = SubnetScanResult(devices=devices).model_dump()
with open(self.storage_file, "w", encoding="utf-8") as f: with open(self.storage_file, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False) json.dump(data, f, indent=2, ensure_ascii=False)
@ -199,18 +156,18 @@ class SubnetService:
def __init__(self): def __init__(self):
self.interface = self._get_default_interface() self.interface = self._get_default_interface()
self.network_scanner = NetworkScanner(self.interface) self.network_scanner = NetworkScanner(self.interface)
self.port_scanner = PortScanner(self.interface)
self.device_info = DeviceInfoCollector() self.device_info = DeviceInfoCollector()
self.port_scanner = PortScanner(self.interface, self.network_scanner.arp_cache)
self.storage = DeviceStorage() self.storage = DeviceStorage()
def _get_default_interface(self): def _get_default_interface(self) -> str:
"""Определяет основной сетевой интерфейс."""
try: try:
gateways = netifaces.gateways() gateways = netifaces.gateways()
default_gateway = gateways.get("default", {}).get(netifaces.AF_INET) default_gateway = gateways.get("default", {}).get(netifaces.AF_INET)
if default_gateway: if default_gateway:
return default_gateway[1] return default_gateway[1]
for iface in netifaces.interfaces(): interfaces = netifaces.interfaces()
for iface in interfaces:
if iface != "lo": if iface != "lo":
return iface return iface
return "eth0" return "eth0"
@ -219,63 +176,79 @@ class SubnetService:
return "eth0" return "eth0"
def scan_subnet(self, subnet: str, ports: List[int] = None) -> SubnetScanResult: def scan_subnet(self, subnet: str, ports: List[int] = None) -> SubnetScanResult:
"""Сканирование подсети и результаты."""
start_time = time.time() start_time = time.time()
devices = [] devices = []
now = datetime.now(ZoneInfo("UTC")).isoformat() now = datetime.now(ZoneInfo("UTC")).isoformat()
try: try:
# Загружаем старую карту
saved_devices = self.storage.load_devices() saved_devices = self.storage.load_devices()
# Получаем локальное устройство
local_device = self._get_local_device(subnet) local_device = self._get_local_device(subnet)
if local_device: if local_device:
local_device.status = "online" local_device.status = "online"
local_device.last_seen = now local_device.last_seen = now
devices.append(local_device) devices.append(local_device)
if local_device.mac: if local_device.mac:
saved_devices[normalize_mac(local_device.mac)] = local_device saved_devices[local_device.mac] = local_device
# Сканируем подсеть
discovered_devices = self.network_scanner.scan(subnet) discovered_devices = self.network_scanner.scan(subnet)
with ThreadPoolExecutor(max_workers=20) as executor: # Уменьшено до 20 # Обрабатываем обнаруженные устройства
def process_device(ip_mac: Tuple[str, str]) -> Optional[Device]: def process_device(ip_mac: Tuple[str, str]) -> Optional[Device]:
ip, mac = ip_mac ip, mac = ip_mac
mac = normalize_mac(mac) if local_device and ip == local_device.ip:
if local_device and ip == local_device.ip: return None
return None # Проверяем, есть ли устройство в старой карте
existing_device = saved_devices.get(mac) existing_device = saved_devices.get(mac)
vendor = self.device_info.get_vendor(mac) if existing_device:
if vendor == "Unknown" and existing_device and existing_device.vendor != "Unknown": # Обновляем данные, если MAC совпадает
vendor = existing_device.vendor
logger.debug(f"Использован сохранённый vendor для MAC {mac}: {vendor}")
return Device( return Device(
ip=ip, ip=ip,
mac=mac, mac=mac,
hostname=self.device_info.get_hostname(ip) or ( hostname=self.device_info.get_hostname(ip) or existing_device.hostname,
existing_device.hostname if existing_device else None), vendor=self.device_info.get_vendor(mac),
vendor=vendor, open_ports=self.port_scanner.scan(ip) or existing_device.open_ports,
open_ports=self.port_scanner.scan(ip) or [],
status="online", status="online",
last_seen=now last_seen=now
) )
# Новое устройство
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=now
)
with ThreadPoolExecutor(max_workers=10) as executor:
device_results = executor.map(process_device, discovered_devices) device_results = executor.map(process_device, discovered_devices)
devices.extend([d for d in device_results if d is not None]) devices.extend([d for d in device_results if d is not None])
# Добавляем оффлайн-устройства из старой карты
active_macs = {d.mac for d in devices if d.mac} active_macs = {d.mac for d in devices if d.mac}
for mac, device in saved_devices.items(): for mac, device in saved_devices.items():
if mac not in active_macs: if mac not in active_macs:
device.status = "offline" device.status = "offline"
devices.append(device) devices.append(device)
devices.sort(key=lambda x: int(ipaddress.ip_address(x.ip))) # Сортировка по IP
devices.sort(key=lambda x: int(ipaddress.IPv4Address(x.ip)))
# Сохраняем обновлённую карту
for device in devices: for device in devices:
if device.mac: if device.mac:
saved_devices[device.mac] = device saved_devices[device.mac] = device
self.storage.save_devices(devices) self.storage.save_devices(list(saved_devices.values()))
logger.info( logger.info(
f"Сканирование подсети {subnet} завершено за {time.time() - start_time:.2f} сек, " f"Сканирование подсети {subnet} завершено за {time.time() - start_time:.2f} секунд, "
f"найдено {len(devices)} устройств (активных: {len(active_macs)})" f"найдено {len(devices)} устройств (активных: {len(active_macs)})"
) )
return SubnetScanResult(devices=devices) return SubnetScanResult(devices=devices)
@ -285,22 +258,22 @@ class SubnetService:
return SubnetScanResult(devices=devices) return SubnetScanResult(devices=devices)
def _get_local_device(self, subnet: str) -> Optional[Device]: def _get_local_device(self, subnet: str) -> Optional[Device]:
"""Получает информацию о локальном устройстве."""
now = datetime.now(ZoneInfo("UTC")).isoformat() now = datetime.now(ZoneInfo("UTC")).isoformat()
try: try:
subnet_network = ipaddress.ip_network(subnet, strict=False) subnet_network = ipaddress.ip_network(subnet, strict=False)
for iface in netifaces.interfaces(): interfaces = netifaces.interfaces()
for iface in interfaces:
if iface == "lo": if iface == "lo":
continue continue
addrs = netifaces.ifaddresses(iface) addrs = netifaces.ifaddresses(iface)
if netifaces.AF_INET in addrs and netifaces.AF_LINK in addrs: if netifaces.AF_INET in addrs and netifaces.AF_LINK in addrs:
ip = addrs[netifaces.AF_INET][0]["addr"] ip = addrs[netifaces.AF_INET][0]["addr"]
if ipaddress.ip_address(ip) in subnet_network: if ipaddress.ip_address(ip) in subnet_network:
mac = normalize_mac(addrs[netifaces.AF_LINK][0]["addr"]) mac = addrs[netifaces.AF_LINK][0]["addr"]
vendor = self.device_info.get_vendor(mac) vendor = self.device_info.get_vendor(mac)
hostname = self.device_info.get_hostname(ip) hostname = self.device_info.get_hostname(ip)
open_ports = self.port_scanner.scan(ip) open_ports = self.port_scanner.scan(ip)
logger.debug(f"Локальное устройство: IP={ip}, MAC={mac}, Vendor={vendor}, Ports={open_ports}") logger.debug(f"Локальное устройство: IP={ip}, MAC={mac}, Ports={open_ports}")
return Device( return Device(
ip=ip, ip=ip,
mac=mac, mac=mac,