Перейти к содержимому

Сейчас: внедряю системы ИИ-мониторинга в Steady Control

Daniil Dushin

Network Inventory

Находит устройства в сети офиса или отеля, ведёт их реестр и отмечает новые устройства и опасные открытые сервисы.

Роль
Личный проект — проектирование, код, тесты
Год
2026
Категория
Инфраструктура
Стек
Python 3.12, asyncio, FastAPI, Jinja2
Обложка Network Inventory — сетка сетевых устройств, одно отмечено оранжевым

Проблема

Спросите небольшой отель или офис, что подключено к их сети, и честный ответ обычно — «не знаем». Камеры, регистраторы, принтеры, точки доступа и компьютер, который кто-то несколько лет назад настроил для удалённой поддержки, сидят в одной плоской сети. Новое устройство никто не замечает, и никто не знает, что камера после сброса настроек отвечает по Telnet.

Директива ЕС NIS2 называет учёт активов среди минимальных мер кибербезопасности (статья 21(2)(i)). Она распространяется на средние и крупные компании определённых отраслей, поэтому многие небольшие предприятия в неё не попадают. Но список того, что есть в сети, — всё равно первый шаг любой работы по безопасности. Этот инструмент помогает вести такой список; соответствия NIS2 он сам по себе не обеспечивает.

Решение

  • Сети для сканирования нужно перечислить явно. Разрешены только частные диапазоны, не больше /22, и инструмент никогда никуда не логинится и ничего не взламывает.
  • Скан пингует все адреса параллельно через asyncio, а потом читает ARP-таблицу операционной системы, чтобы узнать MAC-адреса. Так находится даже компьютер, который не отвечает на ping.
  • К каждому живому устройству делается TCP-подключение на короткий список портов (Telnet, FTP, RDP, SMB, RTSP, Winbox…) и обратный DNS-запрос, чтобы узнать имя.
  • По MAC-адресу определяется производитель — из небольшой встроенной выборки списка IEEE. Случайные MAC телефонов и виртуальных машин помечаются как случайные, а не получают неправильного производителя.
  • Устройства отслеживаются по MAC в реестре SQLite, поэтому новый адрес от DHCP — это не новое устройство. Каждый скан сравнивается с предыдущим: новое устройство, пропало, сменился IP, открылся порт.
  • Новое устройство или новый опасный порт — одно сообщение в Telegram за скан. Веб-страница показывает реестр с фильтрами, риски с понятными объяснениями и выгружает CSV и JSON.

Скриншоты

Страница инвентаризации в демо-режиме с итоговыми цифрами, изменениями последнего скана и началом таблицы устройств
Страница инвентаризации в демо-режиме — одно новое устройство, два пропавших, четыре с высоким риском.
Страница камеры с тремя рисками — открыт Telnet, веб-интерфейс без HTTPS, доступен видеопоток — у каждого есть объяснение
Камера после сброса к заводским настройкам, каждый риск объяснён.
Таблица устройств, отфильтрованная по устройствам хотя бы с одним риском, со значками рисков, производителями, владельцами и местами установки
Все устройства хотя бы с одним риском.

Технологии

  • Python 3.12
  • asyncio
  • FastAPI
  • Jinja2
  • SQLite
  • aiogram 3
  • pytest

Поиск устройств использует системные ping, arp -a и ip neigh, а не «сырые» сокеты, поэтому работает без прав администратора. Их вывод декодируется как ASCII: на русской Windows слова становятся нечитаемыми, но TTL=, IP- и MAC-адреса сохраняются на любом языке — урок из проекта Uptime Monitor. Сопоставление устройств и сравнение сканов — чистые функции, а каждый скан сохраняется одной транзакцией.

Фрагмент кода

Сравнение сканов — чистая функция. Порты сравниваются с последним появлением устройства, даже если оно было несколько недель назад, а пропавшим устройство считается, только если этот скан охватывал его сеть:

python
def diff(*, last_known, previous_scan, current, in_scope):
    changes = []
    for key, now in current.items():
        before = last_known.get(key)
        if before is None:
            changes.append(Change(ChangeKind.NEW, key, tuple(sorted(now.ports)), new_ip=now.ip))
            continue
        if key not in previous_scan:
            changes.append(Change(ChangeKind.RETURNED, key, new_ip=now.ip))
        if before.ip != now.ip:
            changes.append(Change(ChangeKind.IP_CHANGED, key, old_ip=before.ip, new_ip=now.ip))
        if opened := now.ports - before.ports:
            changes.append(Change(ChangeKind.PORT_OPENED, key, tuple(sorted(opened))))
        if closed := before.ports - now.ports:
            changes.append(Change(ChangeKind.PORT_CLOSED, key, tuple(sorted(closed))))

    for key in previous_scan:
        before = last_known.get(key)
        if key not in current and before is not None and in_scope(before.ip):
            changes.append(Change(ChangeKind.GONE, key, old_ip=before.ip))
    return changes

Баг, который я нашёл и исправил

Имена хостов пропадали, когда у многих адресов не было имени

Что сломалось

Прежде чем доверять обратному DNS целую подсеть, я замерил его на своём компьютере с Windows. По отдельности 127.0.0.1 превращался в имя компьютера за 3 мс. В пачке из 61 адреса, где у остальных 60 имени нет, тот же запрос вернулся пустым. В настоящей сети это значит, что устройства теряют имена, как только у многих адресов нет DNS-записи, а это обычная ситуация.

Причина

На Windows запрос для адреса без имени завершается ошибкой только через 4,98 секунды. Таймаут в 2 секунды останавливал лишь ожидание: поток, застрявший в getnameinfo, отменить нельзя, и он оставался занят. Место для параллельного запроса освобождалось по таймауту, поэтому новые запросы вставали в очередь общего пула за занятыми потоками, и их 2 секунды кончались ещё до начала.

Исправление

У обратного DNS теперь свой пул потоков, а место освобождается только когда поток действительно закончил. asyncio.shield не даёт таймауту отменить future — иначе место снова освобождалось бы раньше времени. После исправления та же пачка возвращает имя. Регрессионный тест воспроизводит проблему с подставным медленным резолвером, без сети.

До
try:
    name, _ = await asyncio.wait_for(
        loop.getnameinfo((ip, 0), socket.NI_NAMEREQD), timeout_s
    )
except (TimeoutError, OSError):
    return None
После
await self._slots.acquire()
future = loop.run_in_executor(self._pool, self._resolve, ip)
future.add_done_callback(self._release)  # free the slot only when the thread is done
try:
    name = await asyncio.wait_for(asyncio.shield(future), self._timeout)
except (TimeoutError, OSError, UnicodeError):
    return None

Результат

  • 117 автоматических тестов, включая настоящий вывод ping и arp -a с русской Windows.
  • Проверено на настоящей сети, но только на этом компьютере и домашнем роутере: роутер найден с производителем, именем и открытыми портами за 2–3 секунды.
  • Скан диапазона loopback 127.0.0.0/22 (1022 адреса, все на этом компьютере) занял 22 секунды.
  • Демо-режим моделирует офис из 21 устройства и сканирует его дважды: шесть изменений и одна тревога в Telegram.