Для чего нужна данная статья:
Научиться использованию рекуррентных нейронных сетей (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 часов обедаете и смотрите новости, то это ваш "обычный шаблон трафика". Алгоритм запоминает такие регулярные действия.
Иногда реальных данных мало или их нет, поэтому создают искусственные (фиктивные) данные, похожие на реальные. Это нужно, чтобы протестировать или обучить модель.
Пример:
Если вы учитесь распознавать мошеннические операции по банковским картам, но у вас мало примеров мошенничества, вы можете сгенерировать фиктивные транзакции, похожие на мошеннические, чтобы модель научилась их распознавать.
Многопоточность: отдельные потоки для захвата, сборки и анализа для обеспечения не блокируемой работы.
Здесь программа делит свою работу на три части, каждая из которых выполняется в своём потоке:
- Захват: Получение данных (например, «слушание» сетевого трафика, как микрофон записывает звук).
- Сборка: Преобразование сырых данных в удобный формат (например, из звука — в текст).
- Анализ: Изучение данных, поиск закономерностей (например, поиск ключевых слов в тексте).
Пример:
Представьте, что вы смотрите футбольный матч:
- Один человек (поток) записывает видео с камеры (захват).
- Второй монтирует видео, добавляя титры (сборка).
- Третий анализирует игру, считая голы и фолы (анализ).
Всё происходит одновременно, и результат готов быстрее.
Это значит, что если одна часть программы занята, другие не «замирают» в ожидании, а продолжают работать.
Пример:
Если вы разговариваете по телефону и одновременно печёте пирог, то разговор не останавливается, пока пирог печётся — вы занимаетесь обоими делами параллельно.
В анализе сетевого трафика данные поступают постоянно и в большом объёме. Если бы всё делал один поток, программа могла бы «зависнуть» или работать медленно. А благодаря многопоточности каждая часть работает независимо, и система справляется с нагрузкой эффективнее.
Пример кода
// Cargo.toml dependencies (add these to your Cargo.toml)
// [dependencies]
// pnet = "0.35.0"
// smoltcp = "0.11.0" // For TCP reassembly inspired by blatta-stream
// tch = "0.17.0" // Rust bindings to PyTorch for RNN/LSTM
// ndarray = "0.15.6" // For data manipulation
// crossbeam = "0.8.4" // For multithreading
// parking_lot = "0.12.1" // For synchronization
// rand = "0.8.5" // For random initialization
// chrono = "0.4.38" // For timestamps
// thiserror = "1.0.63"// For error handling
// log = "0.4.22" // For logging
// env_logger = "0.11.5" // Logger setup
use std::env;
use std::thread;
use std::sync::{Arc, Mutex};
use std::collections::HashMap;
use std::time::{Duration, Instant as StdInstant};
use crossbeam::channel::{Sender, Receiver, unbounded};
use parking_lot::RwLock;
use pnet::datalink::{self, NetworkInterface, Config, Channel};
use pnet::packet::ethernet::{EthernetPacket, EtherTypes};
use pnet::packet::ip::{IpNextHeaderProtocols};
use pnet::packet::ipv4::Ipv4Packet;
use pnet::packet::tcp::TcpPacket;
use smoltcp::wire::{Ipv4Address, TcpSeqNumber};
use smoltcp::iface::{SocketSet, Interface};
use smoltcp::socket::{TcpSocket, TcpSocketBuffer};
use tch::{nn, Tensor, Kind, Device, IndexOp, nn::OptimizerConfig};
use ndarray::{Array2, Array1, s};
use rand::Rng;
use chrono::Utc;
use thiserror::Error;
use log::{info, warn, error};
#[derive(Error, Debug)]
enum TrafficAnalysisError {
#[error("Packet capture error: {0}")]
CaptureError(String),
#[error("Reassembly error: {0}")]
ReassemblyError(String),
#[error("ML model error: {0}")]
ModelError(#[from] tch::TchError),
#[error("Feature extraction error: {0}")]
FeatureError(String),
}
#[derive(Clone, Debug)]
struct FlowKey {
src_ip: Ipv4Address,
dst_ip: Ipv4Address,
src_port: u16,
dst_port: u16,
}
#[derive(Clone, Debug)]
struct Flow {
packets: Vec<(Vec<u8>, StdInstant)>, // Payload and timestamp
seq: TcpSeqNumber,
ack: TcpSeqNumber,
last_activity: StdInstant,
}
struct TrafficAnalyzer {
flows: Arc<RwLock<HashMap<FlowKey, Flow>>>,
model: Arc<LstmAutoencoder>,
anomaly_threshold: f64,
device: Device,
}
struct LstmAutoencoder {
vs: nn::VarStore,
lstm: nn::LSTM,
linear: nn::Linear,
}
impl LstmAutoencoder {
fn new(input_size: i64, hidden_size: i64, num_layers: i64, output_size: i64, device: Device) -> Self {
let vs = nn::VarStore::new(device);
let root = vs.root();
let lstm = nn::lstm(&root / "lstm", input_size, hidden_size, nn::RnnConfig {
num_layers,
bidirectional: false,
batch_first: true,
..Default::default()
});
let linear = nn::linear(&root / "linear", hidden_size, output_size, Default::default());
Self { vs, lstm, linear }
}
fn forward(&self, input: &Tensor) -> Tensor {
let (output, _) = self.lstm.forward_t(input, None, false);
let last_hidden = output.i((.., -1, ..)); // Last time step
self.linear.forward(&last_hidden)
}
fn train(&mut self, sequences: &[Array2<f32>], epochs: usize, lr: f64) -> Result<(), TrafficAnalysisError> {
let mut opt = nn::Adam::default().build(&self.vs, lr)?;
for epoch in 0..epochs {
let mut total_loss = 0.0;
for seq in sequences {
let input_tensor = Tensor::from_slice2(seq.view()).to_device(self.vs.device());
let target = input_tensor.copy();
let reconstructed = self.forward(&input_tensor);
let loss = reconstructed.mse_loss(&target, tch::Reduction::Mean);
opt.backward_step(&loss);
total_loss += f64::from(loss);
}
info!("Epoch {}: Loss {}", epoch, total_loss / sequences.len() as f64);
}
Ok(())
}
fn detect_anomaly(&self, sequence: &Array2<f32>) -> bool {
let input_tensor = Tensor::from_slice2(sequence.view()).to_device(self.vs.device());
let reconstructed = self.forward(&input_tensor);
let error = reconstructed.sub(&input_tensor).pow_tensor_scalar(2).mean(Kind::Float);
f64::from(error) > 0.1 // Threshold, adjust based on training
}
}
fn capture_packets(interface_name: &str, tx: Sender<(FlowKey, Vec<u8>, StdInstant)>) -> Result<(), TrafficAnalysisError> {
let interfaces = datalink::interfaces();
let interface = interfaces.into_iter()
.find(|iface: &NetworkInterface| iface.name == interface_name)
.ok_or(TrafficAnalysisError::CaptureError(format!("Interface {} not found", interface_name)))?;
let config = Config {
read_timeout: Some(Duration::from_millis(10)),
..Default::default()
};
let (_, mut rx) = match datalink::channel(&interface, config) {
Ok(Channel::Ethernet(tx, rx)) => (tx, rx),
Ok(_) => return Err(TrafficAnalysisError::CaptureError("Unsupported channel type".into())),
Err(e) => return Err(TrafficAnalysisError::CaptureError(e.to_string())),
};
loop {
match rx.next() {
Ok(packet) => {
if let Some(eth) = EthernetPacket::new(packet) {
if eth.get_ethertype() == EtherTypes::Ipv4 {
if let Some(ipv4) = Ipv4Packet::new(eth.payload()) {
if ipv4.get_next_level_protocol() == IpNextHeaderProtocols::Tcp {
if let Some(tcp) = TcpPacket::new(ipv4.payload()) {
let key = FlowKey {
src_ip: Ipv4Address::from(ipv4.get_source().octets()),
dst_ip: Ipv4Address::from(ipv4.get_destination().octets()),
src_port: tcp.get_source(),
dst_port: tcp.get_destination(),
};
let payload = tcp.payload().to_vec();
if !payload.is_empty() {
tx.send((key, payload, StdInstant::now()))
.map_err(|e| TrafficAnalysisError::CaptureError(e.to_string()))?;
}
}
}
}
}
}
}
Err(e) => warn!("Packet receive error: {}", e),
}
}
}
fn reassemble_flows(rx: Receiver<(FlowKey, Vec<u8>, StdInstant)>, flows: Arc<RwLock<HashMap<FlowKey, Flow>>>, idle_timeout: Duration) {
let mut last_check = StdInstant::now();
loop {
if let Ok((key, payload, timestamp)) = rx.try_recv() {
let mut flows_guard = flows.write();
let flow = flows_guard.entry(key.clone()).or_insert_with(|| Flow {
packets: Vec::new(),
seq: TcpSeqNumber(0), // Simplified, in real use track seq/ack
ack: TcpSeqNumber(0),
last_activity: timestamp,
});
flow.packets.push((payload, timestamp));
flow.last_activity = timestamp;
}
if StdInstant::now() - last_check > Duration::from_secs(10) {
let mut flows_guard = flows.write();
flows_guard.retain(|_, flow| StdInstant::now() - flow.last_activity < idle_timeout);
last_check = StdInstant::now();
}
thread::sleep(Duration::from_millis(1));
}
}
fn extract_features(flow: &Flow) -> Result<Array2<f32>, TrafficAnalysisError> {
// Complex feature extraction: packet sizes, inter-arrival times, entropy, etc.
let mut sizes = Vec::new();
let mut intervals = Vec::new();
let mut prev_time = flow.packets.first().map(|p| p.1).unwrap_or(StdInstant::now());
for (payload, time) in &flow.packets {
sizes.push(payload.len() as f32);
let interval = time.duration_since(prev_time).as_secs_f32();
intervals.push(interval);
prev_time = *time;
}
// Pad to fixed sequence length, say 50
let seq_len = 50;
let mut features = Array2::zeros((seq_len, 2)); // size and interval
for i in 0..sizes.len().min(seq_len) {
features[[i, 0]] = sizes[i] / 1500.0; // Normalize
features[[i, 1]] = intervals[i].min(1.0); // Cap interval
}
// Add more features: entropy
let mut entropy = 0.0;
if !flow.packets.is_empty() {
let all_payload: Vec<u8> = flow.packets.iter().flat_map(|p| p.0.clone()).collect();
let mut counts = [0u32; 256];
for &byte in &all_payload {
counts[byte as usize] += 1;
}
let total = all_payload.len() as f32;
entropy = counts.iter().filter(|&&c| c > 0).map(|&c| {
let p = c as f32 / total;
-p * p.log2()
}).sum();
}
// Duplicate entropy across sequence for simplicity
for i in 0..seq_len {
features[[i, 1]] += entropy / 8.0; // Normalize entropy (max 8 for bytes)
}
Ok(features)
}
fn main() -> Result<(), TrafficAnalysisError> {
env_logger::init();
let args: Vec<String> = env::args().collect();
if args.len() < 2 {
return Err(TrafficAnalysisError::CaptureError("Usage: cargo run <interface>".into()));
}
let interface = &args[1];
let flows = Arc::new(RwLock::new(HashMap::new()));
let (tx, rx) = unbounded();
// Capture thread
let capture_flows = flows.clone();
let capture_handle = thread::spawn(move || capture_packets(interface, tx).unwrap_or_else(|e| error!("Capture error: {}", e)));
// Reassembly thread
let reassembly_flows = flows.clone();
let reassembly_handle = thread::spawn(move || reassemble_flows(rx, reassembly_flows, Duration::from_secs(120)));
// ML model setup
let device = if tch::Cuda::is_available() { Device::Cuda(0) } else { Device::Cpu };
let mut model = LstmAutoencoder::new(2, 64, 2, 2, device); // input: 2 features, hidden 64, 2 layers
// Simulate training with dummy data (in real: collect normal traffic)
let mut rng = rand::thread_rng();
let mut train_sequences = Vec::new();
for _ in 0..100 {
let mut seq = Array2::zeros((50, 2));
for i in 0..50 {
seq[[i, 0]] = rng.gen_range(0.0..1.0);
seq[[i, 1]] = rng.gen_range(0.0..1.0);
}
train_sequences.push(seq);
}
model.train(&train_sequences, 50, 0.001)?;
// Analysis loop in main thread
loop {
{
let flows_guard = flows.read();
for (key, flow) in flows_guard.iter() {
if flow.packets.len() > 10 { // Process mature flows
match extract_features(flow) {
Ok(features) => {
if model.detect_anomaly(&features) {
info!("Anomaly detected in flow {:?} at {}", key, Utc::now());
} else {
info!("Normal flow {:?} at {}", key, Utc::now());
}
}
Err(e) => warn!("Feature error: {}", e),
}
}
}
}
thread::sleep(Duration::from_secs(5));
}
capture_handle.join().unwrap();
reassembly_handle.join().unwrap();
Ok(())
}
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 — Главная функция
Здесь всё связывается вместе:
- Запускаются потоки для захвата и сборки пакетов
- Создаётся и обучается модель
- Периодически проверяются потоки на аномалии
Пример:
Программа запускается, начинает слушать сеть, обучает модель на нормальном трафике, а потом сообщает, если видит что-то подозрительное.