Автоматизированный сбор и анализ сетевого трафика в реальном времени: использование tcpdump, Wireshark CLI и машинного классификатора в Scikit‑learn (Termux)

Termux — удобная среда для практических задач по анализу сетевого трафика на Android: сбор пакетов, извлечение признаков и последующая классификация с использованием машинного обучения. В этой статье мы соберём конвейер «сбор → разбор → признаки → классификация» для реального времени (в пределах возможностей хост‑системы), применяя tcpdump, CLI‑инструменты Wireshark (tshark) и модели в scikit-learn.

Важно: используйте метод только на сетях, где у вас есть законное право на мониторинг. Всегда обеспечивайте согласие владельцев сети/устройств и соблюдайте требования действующего законодательства РФ, а также внутренние политики организации.

Архитектура решения

Практическая схема конвейера:

1. Сбор трафика: tcpdump записывает пакеты в файлы .pcap (с сегментацией по времени).
2. Извлечение признаков: tshark читает pcap и выгружает структуру/поля в формат CSV/TSV.
3. Подготовка данных: агрегация признаков по окнам, очистка, приведение типов.
4. Классификация: модель scikit-learn (например, LogisticRegression / RandomForest) предсказывает класс события/трафика.
5. Повтор: цикл запускается по мере появления новых сегментов.

С точки зрения «реального времени» разумнее говорить о псевдореальном времени: мы обрабатываем трафик блоками (например, каждые 10–30 секунд), что типично для полевых исследований и мониторинга.

Подготовка Termux

Начните с обновления и установки пакетов. В зависимости от устройства и политик Termux, названия пакетов могут незначительно отличаться, но логика сохраняется.

pkg update
pkg upgrade -y
pkg install -y python clang ncurses-utils git tcpdump tshark

Далее поставим Python‑зависимости для анализа и машинного обучения.

pip install --upgrade pip
pip install pandas numpy scikit-learn joblib

Проверьте наличие нужных команд:

which tcpdump
which tshark
python -V
pip show scikit-learn | head

Сбор трафика с tcpdump (сегментация)

Для автоматизированной обработки удобнее писать пакеты в серию файлов, например, каждые 15 секунд. Так вы получите поток «снимков», который можно анализировать по очереди.

Создайте рабочую папку:

mkdir -p ~/netlab/pcap ~/netlab/features ~/netlab/models

Пример запуска tcpdump с сегментацией:

tcpdump -i any -s 0 -nn -tttt 
  -w ~/netlab/pcap/cap_%Y%m%d_%H%M%S_%s.pcap 
  -G 15 -W 0 

Пояснения:

• -i any — попытка мониторинга на всех доступных интерфейсах.
• -s 0 — захват полного пакета.
• -nn — отказ от DNS/имён, быстрее и предсказуемее.
• -tttt — удобный формат времени в выводе.
• -G 15 — файл каждые 15 секунд.
• -W 0 — без ограничения по числу файлов.

Практический совет: если устройство перегружено или файлов получается слишком много, увеличьте окно сегментации (например, до 30/60 секунд) или ограничьте фильтром.

Например, чтобы анализировать только TCP:

tcpdump -i any -s 0 -nn -tttt 
  -w ~/netlab/pcap/cap_%Y%m%d_%H%M%S_%s.pcap 
  -G 15 -W 0 'tcp'

Разбор pcap через Wireshark CLI (tshark)

tshark позволяет извлекать поля из пакетов без открытия GUI. Для автоматизации обычно выбирают несколько полезных полей и агрегируют их.

Пример: выгрузка базовых полей TCP/UDP. Создадим скрипт-помощник (вручную, в nano или через редактор Termux):

cat > ~/netlab/extract_features.py << 'PY'
import sys
import pandas as pd
import subprocess

pcap_path = sys.argv[1]
out_path = sys.argv[2]

# Пример набора полей (можно расширять под свою задачу)
fields = [
    "frame.time_epoch",
    "ip.src",
    "ip.dst",
    "ip.proto",
    "tcp.srcport",
    "tcp.dstport",
    "udp.srcport",
    "udp.dstport",
    "frame.len"
]

# Список колонок, которые будет отдавать tshark
field_args = []
for f in fields:
    field_args += ["-e", f]

cmd = ["tshark", "-r", pcap_path, "-T", "fields"] + field_args + [
    "-E", "separator=	",
    "-E", "occurrence=f",
    "-E", "quote=d",
]

# Запускаем tshark
proc = subprocess.run(cmd, capture_output=True, text=True)
if proc.returncode != 0:
    raise RuntimeError(proc.stderr.strip() or "tshark failed")

lines = [ln for ln in proc.stdout.splitlines() if ln.strip()]
if not lines:
    pd.DataFrame(columns=fields).to_csv(out_path, sep="\t", index=False)
    sys.exit(0)

df = pd.read_csv(pd.io.common.StringIO("
".join(lines)), sep="\t", header=None)
df.columns = fields

# Признаки-агрегации по временным окнам можно делать здесь,
# но в простом примере отдадим "сырые" строки.
df.to_csv(out_path, sep="\t", index=False)
PY

Проверим на одном файле. Сначала дождитесь появления pcap, затем выполните:

pcap_file="$(ls -t ~/netlab/pcap/.pcap | head -n 1)"
feat_file=~/netlab/features/"$(basename "$pcap_file" .pcap)".tsv

python ~/netlab/extract_features.py "$pcap_file" "$feat_file"
ls -lh "$feat_file"
head -n 5 "$feat_file"

Превращаем выгрузку в признаки для классификации

Модели машинного обучения обычно требуют фиксированный набор числовых признаков на «объект» (например, на сегмент/окно). В нашем случае объектом может быть один .pcap файл (15 секунд) или окно агрегации.

Ниже — пример признаков на уровне сегмента: количество пакетов, суммарная длина, доля протоколов, распределения по портам (в упрощённом виде).

cat > ~/netlab/build_dataset.py << 'PY'
import glob
import os
import pandas as pd

features_dir = os.path.expanduser("~/netlab/features")
out_csv = os.path.expanduser("~/netlab/dataset.csv")

rows = []
for path in sorted(glob.glob(os.path.join(features_dir, ".tsv"))):
    df = pd.read_csv(path, sep="	")
    if df.empty:
        continue

    # Базовые признаки сегмента
    packets = len(df)
    bytes_total = df["frame.len"].sum() if "frame.len" in df.columns else 0

    # ip.proto может быть пустым/NaN — обработаем
    proto_counts = df["ip.proto"].fillna("NA").value_counts()
    # Упрощим: возьмём несколько частот протоколов
    proto_tcp = float(proto_counts.get("6", 0))  # 6 = TCP
    proto_udp = float(proto_counts.get("17", 0)) # 17 = UDP
    proto_other = float(packets - proto_tcp - proto_udp) if packets else 0

    # Порты: аккуратная агрегация (можно усложнять)
    # Если это TCP, будут tcp.. Если UDP — udp.
    dst_tcp = df["tcp.dstport"].dropna()
    dst_udp = df["udp.dstport"].dropna()

    # Топ-1 порт по встречаемости (как категориальный -> число можно заменить хеш/индексом)
    dst_port_top = None
    if len(dst_tcp) > 0:
        dst_port_top = dst_tcp.value_counts().index[0]
    elif len(dst_udp) > 0:
        dst_port_top = dst_udp.value_counts().index[0]

    rows.append({
        "pcap_id": os.path.basename(path).replace(".tsv",""),
        "packets": packets,
        "bytes_total": bytes_total,
        "proto_tcp_count": proto_tcp,
        "proto_udp_count": proto_udp,
        "proto_other_count": proto_other,
        # Приведём топ-порт к числу (если None — 0)
        "dst_port_top": float(dst_port_top) if dst_port_top is not None else 0.0,
    })

dataset = pd.DataFrame(rows)
dataset.to_csv(out_csv, index=False)
print("Dataset saved to:", out_csv, "rows:", len(dataset))
PY

Важно: для обучения нужен целевой признак (лейбл). Его можно получить разными законными способами: по заранее известным событиям (например, «контрольный тест» vs «фон» на вашей собственной сети), по журналам вашей системы, либо по ручной разметке сегментов. В статье мы оставим место для label и покажем общий подход.

Обучение классификатора в scikit-learn

Предположим, что у вас уже есть CSV с колонкой label (числовой класс). Тогда обучение будет примерно таким.

cat > ~/netlab/train_model.py << 'PY'
import os
import pandas as pd
import joblib
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LogisticRegression

dataset_path = os.path.expanduser("~/netlab/dataset_with_labels.csv")
model_path = os.path.expanduser("~/netlab/models/model.joblib")

df = pd.read_csv(dataset_path)

if "label" not in df.columns:
    raise ValueError("dataset_with_labels.csv must contain 'label' column")

# Выбираем числовые признаки
feature_cols = [c for c in df.columns if c not in ("pcap_id", "label")]

X = df[feature_cols].fillna(0.0)
y = df["label"]

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.25, random_state=42, stratify=y if len(set(y)) > 1 else None
)

clf = Pipeline([
    ("scaler", StandardScaler()),
    ("lr", LogisticRegression(max_iter=2000))
])

clf.fit(X_train, y_train)
score = clf.score(X_test, y_test)
print("Test accuracy:", score)

joblib.dump({"model": clf, "features": feature_cols}, model_path)
print("Model saved to:", model_path)
PY

Условный запуск:

python ~/netlab/train_model.py

Если вы хотите использовать более гибкие модели (например, RandomForest), замените LogisticRegression на RandomForestClassifier и при необходимости добавьте настройку гиперпараметров.

Онлайн-оценка для новых сегментов (псевдореальное время)

Чтобы классифицировать каждый появившийся pcap, удобнее сделать цикл: замечаем новый файл, извлекаем признаки, строим объект (в нашем примере — агрегацию на уровне сегмента), затем делаем predict.

cat > ~/netlab/stream_classify.sh << 'SH'
#!/data/data/com.termux/files/usr/bin/bash
set -e

PCAP_DIR="$HOME/netlab/pcap"
FEAT_DIR="$HOME/netlab/features"
MODEL_PATH="$HOME/netlab/models/model.joblib"

mkdir -p "$FEAT_DIR"

# Запомним, какие файлы уже обработали
STATE_FILE="$HOME/netlab/state.txt"
touch "$STATE_FILE"

python - << 'PY'
import time, os, glob
import pandas as pd
import joblib

model_bundle = joblib.load(os.path.expanduser("~/netlab/models/model.joblib"))
model = model_bundle["model"]
feature_cols = model_bundle["features"]

pcap_dir = os.path.expanduser("~/netlab/pcap")
feat_dir = os.path.expanduser("~/netlab/features")
state_file = os.path.expanduser("~/netlab/state.txt")

extractor = os.path.expanduser("~/netlab/extract_features.py")

processed = set()
if os.path.exists(state_file):
    with open(state_file, "r", encoding="utf-8") as f:
        for ln in f:
            ln = ln.strip()
            if ln:
                processed.add(ln)

def build_segment_features(tsv_path: str):
    df = pd.read_csv(tsv_path, sep="	")
    if df.empty:
        return None
    packets = len(df)
    bytes_total = df["frame.len"].sum() if "frame.len" in df.columns else 0
    proto_counts = df["ip.proto"].fillna("NA").value_counts()
    proto_tcp = float(proto_counts.get("6", 0))
    proto_udp = float(proto_counts.get("17", 0))
    proto_other = float(packets - proto_tcp - proto_udp) if packets else 0
    dst_tcp = df["tcp.dstport"].dropna()
    dst_udp = df["udp.dstport"].dropna()
    dst_port_top = None
    if len(dst_tcp) > 0:
        dst_port_top = dst_tcp.value_counts().index[0]
    elif len(dst_udp) > 0:
        dst_port_top = dst_udp.value_counts().index[0]
    row = {
        "packets": packets,
        "bytes_total": bytes_total,
        "proto_tcp_count": proto_tcp,
        "proto_udp_count": proto_udp,
        "proto_other_count": proto_other,
        "dst_port_top": float(dst_port_top) if dst_port_top is not None else 0.0,
    }
    # привести к нужному набору
    X = pd.DataFrame([row])
    for c in feature_cols:
        if c not in X.columns:
            X[c] = 0.0
    X = X[feature_cols].fillna(0.0)
    return X

while True:
    pcaps = sorted(glob.glob(os.path.join(pcap_dir, "*.pcap")), key=os.path.getmtime)
    if not pcaps:
        time.sleep(1)
        continue

    for pcap_path in pcaps[-10:]:  # не сканируем всё без необходимости
        base = os.path.basename(pcap_path).replace(".pcap", "")
        if base in processed:
            continue

        feat_path = os.path.join(feat_dir, base + ".tsv")

        # 1) извлечь признаки
        cmd = ["python", extractor, pcap_path, feat_path]
        import subprocess
        r = subprocess.run(cmd, capture_output=True, text=True)
        if r.returncode != 0:
            print("extract_features failed for", base)
            print(r.stderr.strip())
            continue

        # 2) построить сегментные признаки и классифицировать
        X = build_segment_features(feat_path)
        if X is None:
            processed.add(base)
            with open(state_file, "a", encoding="utf-8") as f:
                f.write(base + "
")
            continue

        pred = model.predict(X)[0]
        # pred — номер класса. Для человека можно сделать маппинг label->имя
        print("Segment:", base, "Predicted label:", pred)

        processed.add(base)
        with open(state_file, "a", encoding="utf-8") as f:
            f.write(base + "
")

    time.sleep(2)
PY
SH

Сделайте скрипт исполняемым и запустите:

chmod +x ~/netlab/stream_classify.sh
~/netlab/stream_classify.sh

Если вы хотите ограничить фильтры (например, только внутренний адрес в вашей локальной сети), то корректнее всего применять фильтры на уровне tcpdump (или отдельно при tshark), чтобы уменьшить шум и объём данных.

Фильтрация и локальный контур (VPN как инструмент для локальной сети)

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

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

Типовые ошибки и как их избежать

• Нет данных в pcap: проверьте интерфейс (-i any vs конкретный), права на захват и фильтр.
• tshark возвращает ошибку: убедитесь, что tshark установлен и поля существуют в текущей версии.
• Пустые/битые TSV: проверьте кодировку и формат разделителя, при необходимости укажите -E separator= .
• Модель предсказывает «всё одинаковое»: значит, признаки слабые или данные сильно несбалансированы — добавьте информативные признаки и/или примените class weights.
• Недостаток лейблов: без корректной разметки машинное обучение превращается в угадывание; начните с контролируемых сценариев в своей сети.

Заключение

Мы собрали практичный конвейер для Termux: tcpdump сегментирует и сохраняет трафик в .pcap, tshark извлекает поля в удобный табличный формат, а затем scikit-learn обучает модель и позволяет классифицировать новые сегменты в режиме «псевдореального времени». Такой подход хорошо подходит для учебных лабораторных работ, мониторинга в своей инфраструктуре и задач аналитики, где критична повторяемость и автоматизация.

Если вам нужно развернуть стабильный контур мониторинга, разработать схему разметки и подобрать признаки под вашу конкретную задачу (детектирование сценариев, аномалий или отдельных классов трафика), обратитесь в РыбинскЛАБ — мы поможем спроектировать, внедрить и поддерживать решения для анализа сетевого трафика на базе Termux и инструментов CLI.