Анализ сетевого трафика на Rust ML

Для чего нужна данная статья:

Научиться использованию рекуррентных нейронных сетей (RNN) и LSTM для обнаружения аномалий с использованием ML. Написать анализатор последовательностей сетевых пакетов для выявления низкочастотных атак в HTTP-трафике.

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

Пример:
Представьте, что вы читаете книгу. Чтобы понять смысл предложения, вам нужно помнить, что было написано раньше. RNN работает похоже: она "запоминает" предыдущие элементы последовательности, чтобы лучше анализировать текущий.

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

Если вы читаете длинный рассказ, LSTM поможет не забыть ключевые моменты из начала, даже если вы дошли до конца. В анализе трафика LSTM может заметить, что необычная активность началась несколько часов назад, а не только сейчас.

Низкочастотные атаки в HTTP-трафике - это такие атаки, которые происходят редко и незаметно, но могут нанести вред. Например, кто-то медленно и аккуратно пытается взломать систему, чтобы не вызвать подозрений.

Представьте, что вор пытается взломать сейф. Вместо того чтобы бить по нему молотком (что сразу заметят), он аккуратно подбирает код, пробуя по одной цифре в день. Низкочастотная атака — это как раз такой аккуратный и медленный подбор.

Анализатор последовательностей сетевых пакетов

Захват пакетов в реальном времени: использует pnet для анализа пакетов с интерфейсов Ethernet, фильтруя трафик IPv4/TCP.

Это процесс "прослушивания" и записи всех данных (пакетов), которые передаются по сети прямо сейчас, в момент захвата.

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

pnet - Это библиотека (набор готовых инструментов) для языка программирования Rust, которая помогает работать с сетевыми пакетами.

Если вы строите дом, то вместо того, чтобы делать каждый гвоздь и доску самому, вы покупаете готовые материалы в магазине. pnet — это такой "магазин" для программистов, которые хотят быстро и удобно работать с сетевым трафиком.

Анализ пакетов с интерфейсов Ethernet - программа "слушает" данные, которые проходят через сетевую карту компьютера.

Если ваш компьютер подключён к роутеру кабелем, то программа будет анализировать все данные, которые идут по этому кабелю — как от вас в интернет, так и обратно.

Представьте, что вы слушаете радио и ловите только песни на русском языке, игнорируя всё остальное. Здесь программа ловит только пакеты с адресами IPv4 и протоколом TCP, игнорируя, например, пакеты с видео или другими типами данных.

Сборка TCP-потока: управляет потоками с помощью отслеживания последовательностей и подтверждений, обрабатывает тайм-ауты бездействия для очистки устаревших сеансов.

Сборка TCP-потока — это процесс, при котором все пакеты собираются в правильном порядке, чтобы получить целостные данные (например, страницу сайта, картинку или видео).

Пример:
Представьте, что вы получаете письмо, разрезанное на кусочки. Чтобы прочитать его, нужно сложить все кусочки в правильном порядке. Так же и с TCP: пакеты собираются в единое целое.

Каждый TCP-пакет имеет свой номер (последовательность), чтобы получатель мог понять, в каком порядке их собирать. Когда пакет приходит, получатель отправляет подтверждение (ACK), что пакет получен.

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

Вы заказываете пиццу по телефону. Оператор говорит: "Ваш заказ №1 — пицца, №2 — напиток". Когда курьер привозит пиццу, вы говорите: "Спасибо, пиццу получил!" — это подтверждение.

Если компьютер долго не получает данных от сервера (например, вы закрыли браузер, но соединение не разорвалось), TCP ждёт некоторое время (тайм-аут), а потом закрывает соединение, чтобы не тратить ресурсы.

Вы разговариваете по телефону, но собеседник вдруг замолчал. Через некоторое время вы кладете трубку, чтобы не занимать линию зря.

Расширенная разработка признаков: извлекает нормализованные признаки, такие как размеры пакетов, время между прибытиями и энтропия Шеннона, из полезных данных для машинного обучения.

Это процесс создания новых характеристик (признаков) из сырых данных, чтобы улучшить качество работы моделей машинного обучения.

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

Нормализованные признаки - это признаки, приведённые к единому масштабу, чтобы их было удобнее сравнивать и использовать в моделях.

Если у вас есть размеры пакетов: 100 байт, 1500 байт, 50 байт — их значения сильно отличаются. Нормализация может привести их к шкале от 0 до 1: 0.07, 1.0, 0.03. Теперь все значения сопоставимы, и модель не будет «сбиваться» из-за разницы в масштабах.

Энтропия Шеннона - это мера хаотичности или непредсказуемости данных. Чем выше энтропия, тем менее предсказуемы данные.

Если в сети постоянно идут пакеты одинакового размера с одинаковыми интервалами — энтропия низкая. А если размеры и интервалы всё время разные — энтропия высокая. Это может говорить, например, о необычной активности (вроде атаки).

Модель автокодировщика LSTM: построена с использованием tch-rs (привязка Rust к PyTorch) для анализа последовательностей и обнаружения аномалий на основе ошибок реконструкции.

Автокодировщик — это нейронная сеть, которая учится сжимать и восстанавливать данные. Представьте, что вы фотографируете картину, сжимаете её в маленький файл, а потом восстанавливаете обратно. Если восстановленная картина сильно отличается от оригинала — значит, в данных что-то не так (возможно, аномалия).

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

LSTM — это тип нейронной сети, который хорошо работает с последовательностями данных (например, временными рядами, текстами, видео). Она запоминает важные детали из прошлого и использует их для предсказания будущего.

Представьте, что вы смотрите на график температуры воздуха за неделю. LSTM может заметить, что после трёх жарких дней обычно идёт дождь, и предсказать погоду на завтра.

tch-rs — это библиотека, которая позволяет писать программы на языке Rust, но использовать при этом возможности PyTorch (популярной библиотеки для машинного обучения на Python). То есть, вы получаете скорость и безопасность Rust, но при этом можете использовать мощные инструменты PyTorch.

Представьте, что вы строите дом. PyTorch — это набор инструментов (молоток, пила, гвозди), а tch-rs — это адаптер, который позволяет использовать эти инструменты, даже если вы работаете в другой мастерской (на Rust).

Модель учится восстанавливать нормальные данные. Если ошибка восстановления слишком большая — значит, данные необычные (аномалия).

Допустим, модель обучена на нормальном сетевом трафике офиса. Если вдруг кто-то начинает скачивать огромные файлы в нерабочее время, модель не сможет правильно восстановить этот трафик — ошибка будет большой, и вы поймёте, что что-то не так.

Обучение без учителя: обучение на обычных шаблонах трафика (моделируемых с использованием фиктивных данных в этой версии) для выявления отклонений.

Это метод машинного обучения, при котором модель учится находить закономерности в данных без готовых ответов (меток). То есть, алгоритм сам ищет структуру, кластеры или аномалии в данных, не зная заранее, что именно искать.

Пример:
Представьте, что у вас есть коробка с фруктами: яблоками, бананами и апельсинами. Вы не говорите алгоритму, какие фрукты где лежат, но просите его разложить их по группам. Алгоритм сам заметит, что яблоки круглые и красные, бананы длинные и жёлтые, и разложит их по кучкам — это и есть обучение без учителя.

Обычные шаблоны трафика - это типичное, нормальное поведение данных в сети. Например, как обычно выглядит трафик, когда пользователи заходят на сайт, скачивают файлы, смотрят видео и т.д.

Если вы каждый день в 9 утра проверяете почту, а в 12 часов обедаете и смотрите новости, то это ваш "обычный шаблон трафика". Алгоритм запоминает такие регулярные действия.

Иногда реальных данных мало или их нет, поэтому создают искусственные (фиктивные) данные, похожие на реальные. Это нужно, чтобы протестировать или обучить модель.

Если вы учитесь распознавать мошеннические операции по банковским картам, но у вас мало примеров мошенничества, вы можете сгенерировать фиктивные транзакции, похожие на мошеннические, чтобы модель научилась их распознавать.

Многопоточность: отдельные потоки для захвата, сборки и анализа для обеспечения не блокируемой работы.

Здесь программа делит свою работу на три части, каждая из которых выполняется в своём потоке:

• Захват: Получение данных (например, «слушание» сетевого трафика, как микрофон записывает звук).
• Сборка: Преобразование сырых данных в удобный формат (например, из звука — в текст).
• Анализ: Изучение данных, поиск закономерностей (например, поиск ключевых слов в тексте).

Пример:
Представьте, что вы смотрите футбольный матч:

• Один человек (поток) записывает видео с камеры (захват).
• Второй монтирует видео, добавляя титры (сборка).
• Третий анализирует игру, считая голы и фолы (анализ).
  Всё происходит одновременно, и результат готов быстрее.

Это значит, что если одна часть программы занята, другие не «замирают» в ожидании, а продолжают работать.

Если вы разговариваете по телефону и одновременно печёте пирог, то разговор не останавливается, пока пирог печётся — вы занимаетесь обоими делами параллельно.

В анализе сетевого трафика данные поступают постоянно и в большом объёме. Если бы всё делал один поток, программа могла бы «зависнуть» или работать медленно. А благодаря многопоточности каждая часть работает независимо, и система справляется с нагрузкой эффективнее.

FlowKey — Ключ потока

Это структура, которая однозначно идентифицирует сетевой поток (например, соединение между двумя компьютерами). Она содержит:

• IP-адреса источника и получателя
• Порты источника и получателя

Пример:
Если компьютер с IP 192.168.1.1 и портом 1234 отправляет данные на 192.168.1.2 и порт 80, то FlowKey будет хранить эти четыре значения.

Flow — Сетевой поток

Это структура, которая хранит все пакеты, относящиеся к одному потоку, а также дополнительную информацию:

• Вектор пакетов (данные и время получения)
• Последние номера последовательностей (для TCP)
• Время последней активности

Пример:
Если вы скачиваете файл, все пакеты этого скачивания будут собраны в одном Flow.

TrafficAnalyzer — Анализатор трафика

Это основная структура, которая управляет:

• Всеми потоками (flows)
• Моделью машинного обучения (model)
• Порогом аномалии (anomaly_threshold)

Пример:
Анализатор следит за всеми потоками и использует модель, чтобы определить, есть ли среди них подозрительные (аномалии).

LstmAutoencoder — Модель машинного обучения

Это нейронная сеть, которая обучается на "нормальном" трафике и затем может обнаруживать аномалии.

• LSTM — тип нейронной сети, который хорошо работает с последовательностями (например, временными рядами).
• Autoencoder — сеть, которая учится восстанавливать входные данные. Если ошибка восстановления велика — значит, данные аномальные.

Пример:
Если модель обучена на нормальном трафике (например, обычный веб-серфинг), а потом видит трафик от вируса, она заметит, что он сильно отличается, и сообщит об аномалии.

capture_packets — Захват пакетов

Функция, которая слушает сетевой интерфейс (например, eth0) и перехватывает все пакеты. Она отправляет их в канал (tx) для дальнейшей обработки.

Пример:
Если вы запустили программу на интерфейсе eth0, она будет ловить все пакеты, которые проходят через этот интерфейс.

reassemble_flows — Сборка потоков

Функция, которая собирает пакеты в потоки по ключу (FlowKey). Она удаляет потоки, которые давно не активны.

Пример:
Если вы скачиваете файл, все пакеты этого скачивания будут собраны в один поток, даже если они пришли не подряд.

extract_features — Извлечение признаков

Функция, которая из потока извлекает числовые признаки для модели. Например:

• Размеры пакетов
• Временные интервалы между пакетами
• Энтропия (мера хаотичности данных)

Пример:
Если поток состоит из пакетов размером 100, 200, 150 байт, а интервалы между ними 0.1, 0.3, 0.2 секунды, то функция преобразует это в массив чисел для модели.

main — Главная функция

Здесь всё связывается вместе:

• Запускаются потоки для захвата и сборки пакетов
• Создаётся и обучается модель
• Периодически проверяются потоки на аномалии

Пример:
Программа запускается, начинает слушать сеть, обучает модель на нормальном трафике, а потом сообщает, если видит что-то подозрительное.

Общая архитектура

Сервер предоставляет три API-эндпоинта (все под префиксом /api):

• GET /health – проверка работоспособности.
• POST /packet – приём одного сетевого пакета (в формате MobilePacket) от мобильных устройств, анализ и возврат результата.
• WS /ws – WebSocket-соединение для реального времени: клиенты получают уведомления о найденных аномалиях.

Анализ трафика основан на потоках (flows)
– последовательности пакетов между парой (IP, порт). Сервер накапливает
пакеты в потоках, а когда в потоке набирается достаточно пакетов (≥5),
запускает LSTM-автоэнкодер для детекции аномалий. Результат рассылается
всем подключенным WebSocket-клиентам.

Структуры данных (сериализация)

MobilePacket

Содержит информацию о пакете, поступающем от устройства:

• device_id, timestamp, src_ip, dst_ip, src_port, dst_port, protocol, payload_size, payload_hash.

AnalysisResult

Результат анализа потока:

• flow_id – идентификатор потока (IP+порты);
• is_anomaly – является ли поток аномальным;
• confidence – уверенность модели (0..1);
• features – извлечённые признаки (плоский вектор);
• risk_level – строка: "critical", "high", "medium", "suspicious", "low".

ApiResponse<T>

Обёртка для всех ответов API: { success: bool, data: Option<T>, error: Option<String> }.

Управление WebSocket-клиентами

type WebSocketClients = Arc<RwLock<HashMap<Uuid, mpsc::UnboundedSender<warp::ws::Message>>>>;

• WebSocketClients – потокобезопасный асинхронный RwLock (из tokio::sync) вокруг хеш-таблицы, которая каждому клиенту сопоставляет отправитель (UnboundedSender, используемый для отправки сообщений клиенту из любого места программы через канал).
• Для каждого WebSocket-соединения создаётся mpsc::unbounded_channel(), его отправитель сохраняется в clients, а получатель привязывается к WebSocket-потоку. Это позволяет фоновым задачам (например, broadcast_analysis_result) отправлять сообщения клиенту, не имея прямого доступа к сокету.

Анализатор трафика ApiTrafficAnalyzer

Структура хранит:

• flows – коллекция всех текущих потоков (ключ – FlowKey, значение – Flow). Используется parking_lot::RwLock (быстрая блокировка для синхронных операций, так как работа с потоками происходит в основном внутри async-функций, но критическая секция короткая).
• model – LSTM-автоэнкодер (нейросеть).
• anomaly_threshold – порог аномальности.
• device – устройство (CPU/CUDA) для вычислений.
• ws_clients – ссылка на всех WebSocket-клиентов.

process_mobile_packet

Это главный обработчик входящего пакета.

1. Построение ключа потока (FlowKey) – из IP-адресов и портов отправителя/получателя. IP-адреса парсятся и конвертируются во внутренний формат (Ipv4Address – видимо, из библиотеки libpnet или аналогичной).
2. Добавление пакета в поток:
   Блокировка на запись flows.write().
   Если потока с таким ключом нет – создаётся новый Flow, содержащий вектор пакетов (каждый пакет – пара (payload, Instant)), TCP seq/ack (здесь не используются), время последней активности.
   В flow.packets добавляется симулированный payload (нулевые байты длины payload_size). Это упрощение – в реальности надо извлекать реальную полезную нагрузку.
   Обновляется last_activity.
3. Анализ (если набралось ≥5 пакетов):
   Блокировка на чтение flows.read().
   Вызов self.analyze_flow(&flow_key, flow), который:
   Вызывает внешнюю функцию extract_features(flow)
   (здесь не показана) – вероятно, вычисляет статистические признаки
   потока (межпакетные интервалы, размеры, энтропию) и преобразует их в
   двумерный массив (Array2<f32>).
   Передаёт признаки в модель: self.model.detect_anomaly(&features).
   Вычисляет confidence как среднеквадратичное отклонение признаков (упрощённо).
   Определяет risk_level по таблице: если аномалия и confidence > 0.7 → "critical", и т.д.
   Если анализ вернул Some(result), то вызывается broadcast_analysis_result (рассылка через WebSocket) и результат возвращается в HTTP-ответе.
4. Если пакетов меньше 5, возвращается заглушка (аномалия = false, confidence = 0, risk_level = "low").

Важно: анализ выполняется синхронно внутри async-функции, но блокировки держатся коротко. Модель (LstmAutoencoder) должна быть потокобезопасной (обычно Arc + внутренний Mutex или чисто функциональная).

broadcast_analysis_result

• Сериализует AnalysisResult в JSON (обёрнутый в ApiResponse::success).
• Получает читающую блокировку ws_clients.read().await.
• Для каждого клиента пытается отправить текст через client.send(...). Если канал закрыт – ошибка игнорируется.

API маршруты (warp filters)

with_analyzer и with_clients

Типичные warp-фильтры, которые внедряют в обработчики Arc<ApiTrafficAnalyzer> и WebSocketClients через замыкания.

json_body

Фильтр, который:

• Ограничивает размер тела запроса 16 КБ.
• Десериализует JSON в MobilePacket (используя serde).

handle_packet

Асинхронный обработчик POST-запроса на /api/packet.

• Вызывает analyzer.process_mobile_packet(packet).await.
• В случае успеха возвращает ApiResponse::success(result) (код 200).
• В случае ошибки – ApiResponse::error с текстом ошибки.

handle_websocket

Обработчик WebSocket-апгрейда.

1. Получает warp::ws::Ws (запрос на апгрейд) и список клиентов.
2. Вызывает ws.on_upgrade(...), передавая асинхронную функцию handle_websocket_connection, которая будет вызвана после установки соединения.

handle_websocket_connection

Управляет жизненным циклом одного WebSocket-клиента:

• Разделяет сокет на (sender, receiver) с помощью ws.split().
• Создаёт mpsc::unbounded_channel() – канал для отправки сообщений этому клиенту из внешнего кода.
• Генерирует Uuid для идентификации.
• Сохраняет отправитель (client_sender) в общую clients под этим UUID.
• Отправляет клиенту приветственное сообщение ("Connected to Traffic Analysis API").
• С помощью tokio_stream::wrappers::UnboundedReceiverStream превращает получатель канала в Stream, и запускает задачу tokio::spawn, которая перенаправляет все сообщения из этого канала в WebSocket-отправитель (.forward(client_ws_sender)).
• Затем в цикле читает сообщения от клиента (из client_ws_rcv):
  Если пришло текстовое сообщение – логирует его (можно добавить обработку команд).
  Если пришёл close – выходит из цикла.
  При ошибке – тоже выходит.
• При выходе из цикла удаляет клиента из clients.

Запуск сервера

В main:

1. Инициализация логгера (env_logger).
2. Выбор устройства: Cuda(0) если CUDA доступна, иначе Cpu.
3. Создание модели LstmAutoencoder
   (параметры: входной размер = 2, скрытой размер = 64, 2 слоя, выходной
   размер = 2). Это автоэнкодер для последовательностей длины 50? (видно в
   синтетическом обучении).
4. Попытка загрузить веса из файла model.ot. Если файла нет – генерируется синтетическая обучающая выборка:
   100 последовательностей длины 50, каждая – случайные числа от 0 до 1, размерность признаков = 2.
   Модель обучается 50 эпох с lr=0.001.
   Веса сохраняются.
5. Создаются пустые flows (синхронный PRwLock для быстрого доступа) и ws_clients (асинхронный RwLock).
6. Создаётся ApiTrafficAnalyzer (обёрнут в Arc для разделения между обработчиками).
7. Определяются маршруты: все они группируются под /api, к ним добавляется CORS (любой origin, методы GET/POST/OPTIONS, заголовок content-type).
8. Запуск warp::serve на 0.0.0.0:8080.

Используемые конкурентные примитивы

• std::sync::Arc – для разделения владения анализатором и списком клиентов между несколькими задачами tokio.
• tokio::sync::RwLock – асинхронная блокировка для WebSocketClients. Добавление/удаление клиента и рассылка сообщений требуют ожидания блокировки без блокировки потока (async-await).
• parking_lot::RwLock – синхронная, но очень быстрая блокировка для flows. Критическая секция маленькая (вставка/чтение из HashMap), и асинхронный код в ней не выполняется, поэтому нет риска зависания.
• tokio::sync::mpsc::unbounded_channel – используется для связи между broadcast_analysis_result (или любым другим кодом) и WebSocket-задачей клиента. Неограниченный буфер (но осторожно при высокой нагрузке).
• futures::StreamExt – для удобной работы с UnboundedReceiverStream и метода .forward().

Обработка ошибок и логирование

• Все ошибки (десериализация, ошибки модели, извлечение признаков) логируются с помощью error! / warn! макросов.
• В HTTP-ответах ошибка превращается в ApiResponse::error и возвращается (не паника).
• WebSocket-ошибки просто печатаются в eprintln и разрывают соединение.