Создание распределённого фреймворка для анализа трафика в реальном времени с использованием Zeek, Suricata и Elasticsearch в Termux

Мобильная платформа Android и окружение Termux позволяют развернуть прикладной контур мониторинга сети: сбор сетевых событий, их первичную обработку и агрегацию в хранилище для последующего поиска и визуализации. В этой статье мы рассмотрим концепцию распределённого фреймворка для анализа трафика в реальном времени с использованием Zeek, Suricata и Elasticsearch на узлах, работающих под управлением Termux, с последующей отправкой событий в центральное хранилище.

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

Архитектура распределённого решения

Рекомендуемая архитектура выглядит так:

Узлы сенсоров (Agents) — Termux на Android, где запускаются Zeek и/или Suricata для получения событий (логов/сигналов).
Агрегатор событий — отдельный сервис (может быть на сервере Linux или в контейнере), который принимает данные от агентов и нормализует формат.
Elasticsearch (центральное хранилище) — индексирование событий и быстрый поиск.
Визуализация/поиск — опционально (например, через Kibana), хотя в рамках данной статьи фокус на цепочке Zeek/Suricata → Elasticsearch.

Важно: на Android ограничены типичные сетевые возможности (пакетный захват зависит от прав/конфигурации). Поэтому в реальном проекте чаще всего делают одно из двух:

получают события из уже доступных источников (зеркалирование/агрегация на стороне роутера/хоста, а на Android приходит трафик через разрешённый канал);
или используют режимы/подходы, где данные берутся не “как в ноутбуке”, а через прокси/интеграции (в зависимости от вашего стенда).

Подготовка Termux и базовые требования

Для начала обновите пакеты и установите необходимые инструменты сборки:

pkg update -y
pkg upgrade -y
pkg install -y git curl wget clang make cmake tar xz-utils openssl-tool jq python

Далее зафиксируйте версию Java для Elasticsearch (как правило, совместимая среда Java требуется на машине, где крутится Elasticsearch, а не обязательно на самом Android). На Android Java можно использовать для вспомогательных сервисов, но чаще Elasticsearch размещают на сервере.

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

Установка Elasticsearch и подготовка индексов

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

Для Elasticsearch потребуется:

память и диски под индексы;
настройка сети (доступ агентов только в пределах вашей сети);
создание индексов и шаблонов маппинга под события Zeek/Suricata.

Ниже приведён пример концепции индекса: используйте разные индексы/alias для “zeek” и “suricata”, чтобы упростить аналитику.

# Пример (выполнить на хосте с доступом к Elasticsearch):
# 1) Создать индекс/шаблон можно через API.
# 2) Ниже — упрощенный пример без претензии на идеальный маппинг.

curl -X PUT "http://ELASTICSEARCH_HOST:9200/zeek-000001" -H "Content-Type: application/json" -d '{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0
  }
}'

Для стабильной работы в продакшене используйте ILM (Index Lifecycle Management), шаблоны индексов и строгий маппинг полей. Мы сосредоточимся на “транспортной” части цепочки событий.

Zeek на Termux: роль и формат событий

Zeek — это система анализа сетевого трафика с богатым набором событий уровня “сессия/протокол/состояние”. На практике Zeek требует доступа к сетевым данным (пакетам). На Android это обычно реализуется через заранее подготовленный контур, где сенсор получает данные в разрешённом режиме.

Концептуально задача Zeek для агента:

сгенерировать события в логах (например, TSV/JSON — в зависимости от конфигурации);
отправить эти события в агрегатор или напрямую в Elasticsearch.

Если вы разворачиваете Zeek в виде процесса на узле, ориентируйтесь на лог-вывод и последующую выгрузку. Пример “каркаса” для обработки логов:

# Пример директории проекта на Termux
mkdir -p ~/lab-ids/zeek-logs ~/lab-ids/outputs

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

Suricata на Termux: сигнатуры и алерты

Suricata обеспечивает детекцию по сигнатурам и/или правилам, а также выдаёт алерты с контекстом (IP/порт/протокол/скоринг и т.д.). На Android аналогично важны ограничения по доступу к трафику.

Роль Suricata для агента:

провести детекцию на локальных данных;
сформировать алерты (alert/log);
отправить алерты в центральное хранилище.

Обычно алерты удобнее отправлять в формате JSON (если ваш конвейер это поддерживает). Если на выходе TSV/JSON — приведите к единому “event schema” в агрегаторе.

Конвейер событий: от агента к Elasticsearch

Есть два основных подхода:

Прямой (agent → Elasticsearch): проще в прототипе, но сложнее в контроле схем, ретраев и безопасности.
Через агрегатор (agent → aggregator → Elasticsearch): лучше для нормализации, валидации, дедупликации и управления нагрузкой.

В распределённых системах практичнее второй вариант. Ниже приведён пример концепции простого приемника на Python (на стороне хоста агрегатора), который принимает события от агентов и отправляет их в Elasticsearch.

Пример: простой HTTP приемник (агрегатор)

# На хосте агрегатора (не на самом Termux), например:
python -m pip install --upgrade pip
pip install flask requests

# app.py
from flask import Flask, request, jsonify
import requests

app = Flask(name)

ELASTICSEARCH_URL = "http://ELASTICSEARCH_HOST:9200"
INDEX_ZEEK = "zeek-events"
INDEX_SURICATA = "suricata-events"

@app.post("/ingest")
def ingest():
    payload = request.get_json(force=True)
    source = payload.get("source")  # "zeek" или "suricata"
    doc = payload.get("event")

    if source not in ("zeek", "suricata"):
        return jsonify({"ok": False, "error": "invalid source"}), 400

    index = INDEX_ZEEK if source == "zeek" else INDEX_SURICATA

    # Простой вариант: индексирование единичного документа
    r = requests.post(
        f"{ELASTICSEARCH_URL}/{index}/_doc",
        json=doc,
        timeout=10
    )
    return jsonify({"ok": True, "status": r.status_code})

if name == "main":
    # Для разработки; в продакшене используйте gunicorn/uvicorn и reverse proxy
    app.run(host="0.0.0.0", port=8080)

Запуск:

python app.py

Дальше агенты будут отправлять события на http://AGGREGATOR_HOST:8080/ingest.

Пример отправки событий с Termux в агрегатор

На стороне Termux можно реализовать простой “хвост” логов и отправку новых событий. Ниже пример демонстрирует принцип: вы читаете JSON-строки или приводите записи к JSON и отправляете их в агрегатор.

Подготовка на Termux

pkg install -y python
pip install --upgrade pip
pip install requests

Скрипт отправки событий

# send_events.py
import json
import time
import requests

AGGREGATOR_URL = "http://AGGREGATOR_HOST:8080/ingest"
SOURCE = "zeek"  # или "suricata"

def send_event(event):
    payload = {"source": SOURCE, "event": event}
    r = requests.post(AGGREGATOR_URL, json=payload, timeout=10)
    r.raise_for_status()

# Пример: отправка событий из файла построчно (каждая строка — JSON)
LOG_FILE = "/sdcard/Download/zeek-events.jsonl"  # подставьте свой путь

def main():
    with open(LOG_FILE, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            event = json.loads(line)
            send_event(event)
            time.sleep(0.05)

if name == "main":
    main()

Запуск:

python send_events.py

Для реального “в реальном времени” вместо чтения статического файла используйте tail-подход: отслеживание конца файла, inotify-подход (если доступен в вашей сборке/конфигурации) или интеграцию с тем, как Zeek/Suricata пишут логи.

Нормализация схемы событий (рекомендуемый подход)

Чтобы в Elasticsearch было удобно искать и строить корреляцию, заложите единый набор полей. Например:

timestamp — время события (ISO8601);
source — zeek/suricata;
event_type — тип события;
src.ip, src.port, dst.ip, dst.port;
protocol;
message или details;
hash/flow_id — если у вас есть идентификаторы потоков.

Агрегатор принимает входные события от обоих движков и приводит их к общей структуре. Это снижает “зоопарк” индексов и ускоряет создание дашбордов.

Безопасность и сетевой контур

Для законной эксплуатации важно обеспечить:

доступ к агрегатору и Elasticsearch только из вашей локальной сети (или через ваш безопасный VPN для локальной адресации);
аутентификацию на уровне API приемника (например, токен в заголовке);
ограничение индексации (bulk, rate limiting, ретраи, дедупликация);
шифрование канала при передаче по сети (HTTPS, где применимо).

Для API приемника можно добавить простую проверку токена. Например:

# Идея (фрагмент): проверять header X-API-KEY
# if request.headers.get("X-API-KEY") != EXPECTED_KEY: return 401

Тестирование и отладка

Рекомендуемый план внедрения:

Локальный одиночный узел: поднять Zeek и/или Suricata, проверить генерацию событий и отправку в агрегатор.
Проверка индекса: убедиться, что документы появляются в нужных индексах и поля маппятся корректно.
Нагрузочный прогон: оценить скорость индексации, размер документов и влияние на Elasticsearch.
Добавление второго узла: проверить распределённость и отсутствие коллизий.
Корреляция: сопоставить события Zeek (контекст сессий) и Suricata (сигнатуры/алерты).

Для отладки используйте отдельные индексы/aliases и добавляйте уникальные идентификаторы событий, чтобы легче отслеживать дедупликацию.

Заключение

Распределённый фреймворк анализа трафика в реальном времени на Termux становится практичным, если правильно разделить роли: узлы сенсоров (Zeek/Suricata) генерируют события, агрегатор нормализует их и управляет доставкой, а Elasticsearch обеспечивает быстрый поиск и аналитику. Такой подход масштабируется на несколько Android-узлов и позволяет строить единый “контур наблюдаемости” вашей сети при соблюдении законности и политики безопасности.

Если вам нужен аудит архитектуры, подбор оптимальных индексов/схемы событий, настройка конвейера доставки или помощь в развёртывании на вашей инфраструктуре, обращайтесь в РыбинскЛАБ — мы поможем спроектировать и внедрить решение под ваши требования.