111 lines
No EOL
4.9 KiB
Python
111 lines
No EOL
4.9 KiB
Python
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) |