Wazuh: расширяем функциональность с помощью индексов

Почему индексер, а не Server API

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

Wazuh Server API (порт 55000) — это управление: агенты, группы, конфиги, перезапуск демонов, состояние кластера. Он умеет отдавать инвентарь через syscollector-эндпоинты, но по одному агенту за раз. Собрать через него «все пакеты на всех хостах» технически можно, но практически — это N запросов и полчаса ожидания.

Wazuh Indexer (порт 9200) — это OpenSearch. Здесь лежат сами данные: алерты, состояния, инвентарь. Также в индексере есть агрегации, а один запрос отвечает на вопрос в духе «топ-20 уязвимых пакетов по парку из 3 000 хостов» за пару секунд.

Что вообще есть в индексере

В Wazuh Indexer данные разделены на несколько групп: потоковые события, текущее состояние инфраструктуры, информация об уязвимостях, информация о контроле целостности и служебная информация. Рассмотрим основной набор, актуальный для версии 4.14, в формате небольших шпаргалок.

Данные потока событий

Начнем с индексов, в которых хранится поток событий. Именно с ними чаще всего работают при расследовании инцидентов, поиске подозрительной активности и анализе истории срабатываний правил.

Индекс	Что внутри
wazuh-alerts-4.x-*	Алерты — события, сработавшие по правилам. Основной рабочий индекс
wazuh-archives-4.x-*	Вообще все события, включая не сработавшие ни по одному правилу. По умолчанию выключен

Состояния (stateful-данные, «как есть прямо сейчас»)

Обозначим, что ключевая разница между alerts выше и states, которые рассмотрим далее  — в модели данных. wazuh-alerts-* это append-only поток: каждый алерт, новый документ, индекс режется по дням, а старое удаляется по ISM-политике. wazuh-states-* это срез текущего состояния: документ на сущность (агент + пакет, агент + CVE), обновляется на месте. Поэтому «сколько у меня сейчас критичных CVE» надо спрашивать у states, а «когда мы их обнаружили и что с ними делали» — у alerts.

Индекс	Что внутри
wazuh-states-vulnerabilities-*	Обнаруженные уязвимости по агентам
wazuh-states-inventory-packages-*	Установленные пакеты
wazuh-states-inventory-system-*	ОС, hostname, архитектура
wazuh-states-inventory-hardware-*	CPU, RAM, диски
wazuh-states-inventory-processes-*	Запущенные процессы
wazuh-states-inventory-ports-*	Открытые порты
wazuh-states-inventory-networks-*	Назначенные IPv4/IPv6-адреса
wazuh-states-inventory-interfaces-*	Сетевые интерфейсы и счетчики пакетов
wazuh-states-inventory-protocols-*	Маршруты и протоколы
wazuh-states-inventory-hotfixes-*	Установленные патчи Windows
wazuh-states-inventory-users-*	Учетные записи
wazuh-states-inventory-groups-*	Группы
wazuh-states-inventory-services-*	Системные сервисы
wazuh-states-inventory-browser-extensions-*	Расширения браузеров

Последние четыре индекса (users, groups, services, browser-extensions) приехали только в версии 4.14. Если у вас 4.12, то их там нет.

Служебное

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

Индекс	Что внутри
wazuh-monitoring-*	Слепки статуса агентов (active/disconnected/never_connected), пишет дашборд по расписанию
wazuh-statistics-*	Метрики производительности сервера: принято событий, обработано, дропнуто

Готовим доступ

Наверняка первое, что захочется сделать — взять пользователя admin и не думать. Но не надо. Вам потребуется пять минут на настройку нормального read-only пользователя, чтобы избавиться от лишних рисков.

В дашборде откройте Indexer management → Security → Roles и создайте новую роль (Create role).

• Index permissions: wazuh-alerts-*, wazuh-states-*;
• Permissions: read, indices:data/read/search, indices:data/read/scroll, indices:data/read/point_in_time/*, indices:monitor/*.

Затем перейдите в Internal users → Create user, создайте пользователя с именем py-reader и назначьте ему созданную роль.

Тем, кто любит конфиги, то же самое можно сделать через roles.yml/internal_users.yml в /etc/wazuh-indexer/opensearch-security/ с последующим прогоном securityadmin.sh. Через UI быстрее, но UI-изменения переживают не все сценарии апгрейда, так что для прода yml будет даже лучше.

Теперь о сертификате. Если ставили его из коробки, CA лежит в /etc/wazuh-indexer/certs/root-ca.pem. Скопируйте его на машину, с которой будет запускаться Python-скрипт. Отключать проверку сертификата (verify_certs=False) в проде — плохая идея, но для «посмотреть быстро» сойдет.

Проверяем настройку:

      curl -k -u py-reader:'пароль' https://wazuh-indexer:9200/_cat/indices/wazuh-*?v

Если тут вернулся список — доступ настроен корректно и можно переходить к работе из Python.

Клиент

Для работы из Python ставим официальный клиент OpenSearch — opensearch-py. Несмотря на схожий API, использовать elasticsearch-py здесь не стоит: клиент Elasticsearch 8.x проверяет совместимость с Elastic-кластером и может завершить подключение с ошибкой.

      pip install opensearch-py

Базовое подключение выглядит так:

      import os
from opensearchpy import OpenSearch

client = OpenSearch(
    hosts=[{"host": os.getenv("WZ_HOST", "wazuh-indexer"), "port": 9200}],
    http_auth=(os.environ["WZ_USER"], os.environ["WZ_PASS"]),
    use_ssl=True,
    verify_certs=True,
    ca_certs="/etc/wazuh/root-ca.pem",
    ssl_assert_hostname=False,   # если в сертификате не то CN, что в hosts
    timeout=60,
    max_retries=3,
    retry_on_timeout=True,
)

print(client.info()["version"])

В примере есть два параметра, на которые стоит обратить внимание. ssl_assert_hostname=False — компромисс, но частый: дефолтные сертификаты Wazuh выписаны на имена вроде wazuh-indexer, а ходят к нему обычно по IP или FQDN. Проверку CA при этом мы не выключаем, а именно ее и хочется сохранить. И про таймауты: значение timeout=60 мы ставим не от жадности. Тяжелая агрегация, например, по трехмесячному диапазону легко уезжает за дефолтные 10 секунд, и в итоге клиент отваливается ровно в тот момент, когда индексер уже почти все посчитал.

Что внутри алерта

Прежде чем писать запросы, полезно посмотреть на документ живьем:

      doc = client.search(index="wazuh-alerts-*", body={"size": 1, "query": {"match_all": {}}})
print(doc["hits"]["hits"][0]["_source"])

Скелет типичного алерта выглядит следующим образом:

      {
  "timestamp": "2026-08-02T14:03:11.482+0300",
  "agent": {"id": "004", "name": "web-prod-02", "ip": "10.20.3.14"},
  "manager": {"name": "wazuh-master"},
  "rule": {
    "id": "5710",
    "level": 5,
    "description": "Attempt to login using a non-existent user",
    "groups": ["syslog", "sshd", "authentication_failed"],
    "mitre": {"id": ["T1110"], "tactic": ["Credential Access"],
              "technique": ["Brute Force"]},
    "firedtimes": 3,
    "mail": false
  },
  "decoder": {"name": "sshd"},
  "location": "/var/log/auth.log",
  "full_log": "Aug  2 14:03:11 web-prod-02 sshd[2841]: Invalid user admin from 203.0.113.77",
  "data": {"srcip": "203.0.113.77", "srcport": "51234"},
  "id": "1754140991.9182734"
}

Что стоит держать в голове:

• rule.level — от 0 до 16. Практический водораздел: 7 и более — заслуживает внимания, 12 и более — «бегите смотреть». Все, что ниже 5 — в 95% случаев шум.
• rule.groups — самое недооцененное поле. По нему удобно резать выборки без зависимости от конкретных ID-правил: authentication_failed, web_scan, rootcheck, sca, vulnerability-detector.
• data.* — зона динамического маппинга. Для разных декодеров есть разные поля, и на больших инсталляциях именно она раздувает количество полей в индексе.
• timestamp может приходить со смещением часового пояса, а не в UTC. При работе в Python лучше сразу приводить его к единому формату времени.
• agent.id — строка с ведущими нулями (“004”), а не число. При join’ах с другими источниками это вечная боль.

Схему индекса всегда можно посмотреть напрямую:

      mapping = client.indices.get_mapping(index="wazuh-alerts-*")
# по одному конкретному индексу — читабельнее

Читаем данные: три способа получить документы из Wazuh Indexer

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

1. Обычный search — когда данных мало

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

      from datetime import datetime, timedelta, timezone

since = (datetime.now(timezone.utc) - timedelta(hours=24)).isoformat()

resp = client.search(
    index="wazuh-alerts-*",
    body={
        "size": 100,
        "sort": [{"timestamp": {"order": "desc"}}],
        "query": {
            "bool": {
                "filter": [
                    {"range": {"timestamp": {"gte": since}}},
                    {"range": {"rule.level": {"gte": 10}}},
                ],
                "must_not": [
                    {"terms": {"rule.id": ["5715", "5501", "5502"]}}
                ],
            }
        },
    },
)

for hit in resp["hits"]["hits"]:
    src = hit["_source"]
    print(f"{src['timestamp']}  L{src['rule']['level']:>2}  "
          f"{src['agent']['name']:<20} {src['rule']['description']}")

Два момента, которые экономят нервы.

Для условий отбора используем filter вместо must. В этом режиме фильтровый контекст не считает релевантность и кэшируется. Для «дай все за сутки с уровнем выше N» релевантность бессмысленна, а разница в скорости на больших индексах заметная.

size жестко ограничен параметром index.max_result_window (по умолчанию 10 000 документов). Попытка забрать больше через from/size упрется в ошибку. Что делать — ниже.

2. helpers.scan — когда надо выгрузить все

Если задача стоит уже в обработке тысячи или миллионов документов, использовать обычный search не стоит.

      from opensearchpy import helpers

query = {
    "query": {
        "bool": {
            "filter": [
                {"range": {"timestamp": {"gte": "now-7d"}}},
                {"terms": {"rule.groups": ["authentication_failed"]}},
            ]
        }
    }
}

count = 0
for doc in helpers.scan(
    client,
    index="wazuh-alerts-*",
    query=query,
    size=2000,            # документов за один заход
    scroll="5m",
    request_timeout=120,
    preserve_order=False, # важно: с True скан становится сильно медленнее
):
    count += 1
    # doc["_source"] — тело документа
print(count)

3. search_after — когда нужна возобновляемая выгрузка

Скролл нельзя поставить на паузу и продолжить завтра с того же места, а вот search_after — можно. Курсором служат значения полей сортировки, их достаточно сохранить.

      def iter_alerts(client, gte, lte, page=1000):
    after = None
    body = {
        "size": page,
        "sort": [{"timestamp": "asc"}, {"_id": "asc"}],
        "query": {"bool": {"filter": [
            {"range": {"timestamp": {"gte": gte, "lte": lte}}}
        ]}},
    }
    while True:
        if after:
            body["search_after"] = after
        hits = client.search(index="wazuh-alerts-*", body=body)["hits"]["hits"]
        if not hits:
            return
        yield from hits
        after = hits[-1]["sort"]   # вот его и сохраняем между запусками

Второй ключ сортировки (_id) обязателен: у алертов одинаковый timestamp встречается сплошь и рядом, и без tie-breaker вы получите пропуски и дубли на границах страниц.

Итак:

• search — несколько десятков/сотен результатов для просмотра и анализа;
• helpers.scan — большая одноразовая выгрузка;
• search_after — регулярная обработка, где нужен надежный курсор.

Агрегации — то, ради чего все затевалось

Частая ошибка при работе с OpenSearch — выгружать тысячи документов в Python только для того, чтобы потом посчитать их в pandas. Большинство таких задач лучше отдавать самому Indexer: агрегации выполняются рядом с данными, а по сети передается только результат.

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

Топ атакующих IP за неделю

      resp = client.search(
    index="wazuh-alerts-*",
    body={
        "size": 0,
        "query": {"bool": {"filter": [
            {"range": {"timestamp": {"gte": "now-7d"}}},
            {"exists": {"field": "data.srcip"}},
            {"range": {"rule.level": {"gte": 5}}},
        ]}},
        "aggs": {
            "by_ip": {
                "terms": {"field": "data.srcip", "size": 20,
                          "order": {"_count": "desc"}},
                "aggs": {
                    "targets": {"cardinality": {"field": "agent.id"}},
                    "max_level": {"max": {"field": "rule.level"}},
                },
            }
        },
    },
)

for b in resp["aggregations"]["by_ip"]["buckets"]:
    print(f"{b['key']:<16} {b['doc_count']:>7}  "
          f"хостов: {b['targets']['value']:>3}  "
          f"макс.уровень: {int(b['max_level']['value'])}")

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

Динамика по часам, разложенная по уровням

Еще один типичный сценарий — построить временной ряд: например, увидеть всплески активности и понять, когда именно начался инцидент.

      body = {
    "size": 0,
    "query": {"range": {"timestamp": {"gte": "now-48h"}}},
    "aggs": {
        "over_time": {
            "date_histogram": {
                "field": "timestamp",
                "calendar_interval": "1h",
                "time_zone": "Europe/Moscow",
                "min_doc_count": 0,
            },
            "aggs": {
                "severity": {
                    "range": {
                        "field": "rule.level",
                        "ranges": [
                            {"key": "low", "to": 7},
                            {"key": "medium", "from": 7, "to": 12},
                            {"key": "high", "from": 12},
                        ],
                    }
                }
            },
        }
    },
}

Уязвимости в разрезе критичности и хостов

      body = {
    "size": 0,
    "aggs": {
        "sev": {
            "terms": {"field": "vulnerability.severity", "size": 10},
            "aggs": {
                "hosts": {"cardinality": {"field": "agent.id"}},
                "top_cve": {"terms": {"field": "vulnerability.id", "size": 10}},
            },
        }
    },
}
resp = client.search(index="wazuh-states-vulnerabilities-*", body=body)

Именно такие запросы обычно и нужны на практике: стандартные представления уязвимостей помогают посмотреть состояние конкретного агента или общий список проблем, но не отвечают на вопрос «какие CVE затрагивают больше всего хостов и должны попасть в первую очередь в план исправлений». Такой рейтинг приходится строить самостоятельно через агрегации.

У terms есть ограничение: он хорошо подходит для поиска самых популярных значений, но не для полного перебора большого количества уникальных комбинаций. Если бакетов становится слишком много, большой size увеличивает нагрузку на память координирующей ноды.

Для полного обхода больших наборов данных используется composite-агрегация. Она поддерживает пагинацию через after_key, поэтому ее можно безопасно выполнять порциями:

      def iter_composite(client, index, sources, query=None, size=1000):
    after = None
    while True:
        agg = {"composite": {"size": size, "sources": sources}}
        if after:
            agg["composite"]["after"] = after
        body = {"size": 0, "aggs": {"c": agg}}
        if query:
            body["query"] = query
        res = client.search(index=index, body=body)["aggregations"]["c"]
        if not res["buckets"]:
            return
        yield from res["buckets"]
        after = res.get("after_key")
        if not after:
            return

sources = [
    {"agent": {"terms": {"field": "agent.name"}}},
    {"pkg": {"terms": {"field": "package.name"}}},
]
for b in iter_composite(client, "wazuh-states-vulnerabilities-*", sources):
    print(b["key"], b["doc_count"])

Собираем что-то полезное

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

Еженедельный отчет в Excel

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

      import pandas as pd
from opensearchpy import helpers

rows = []
for doc in helpers.scan(
    client,
    index="wazuh-states-vulnerabilities-*",
    query={"query": {"terms": {"vulnerability.severity": ["Critical", "High"]}}},
    size=2000,
    preserve_order=False,
):
    s = doc["_source"]
    rows.append({
        "host": s["agent"]["name"],
        "agent_id": s["agent"]["id"],
        "cve": s["vulnerability"]["id"],
        "severity": s["vulnerability"]["severity"],
        "cvss": s["vulnerability"].get("score", {}).get("base"),
        "package": s.get("package", {}).get("name"),
        "version": s.get("package", {}).get("version"),
        "detected": s["vulnerability"].get("detected_at"),
    })

df = pd.DataFrame(rows)

# «сколько хостов ждет каждый CVE» — приоритет для патч-менеджмента
prio = (df.groupby(["cve", "severity", "package"])
          .agg(hosts=("host", "nunique"), cvss=("cvss", "max"))
          .reset_index()
          .sort_values(["hosts", "cvss"], ascending=False))

with pd.ExcelWriter("vulns.xlsx", engine="openpyxl") as xl:
    prio.to_excel(xl, sheet_name="Приоритеты", index=False)
    df.to_excel(xl, sheet_name="Детально", index=False)

Всего 20 строк, а на выходе то, за чем обычно ходят к вендору интеграции.

Обратите внимание на .get() с дефолтами. В states-индексах поля вроде package.version или vulnerability.score есть не у всех документов: это зависит от источника детекта. Прямое обращение по ключу рано или поздно уронит скрипт на середине выгрузки.

Свой алертинг с дедупликацией

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

      from datetime import datetime, timedelta, timezone

WINDOW_MIN = 10
THRESHOLD = 50

def brute_force_candidates(client):
    body = {
        "size": 0,
        "query": {"bool": {"filter": [
            {"range": {"timestamp": {"gte": f"now-{WINDOW_MIN}m"}}},
            {"terms": {"rule.groups": ["authentication_failed"]}},
        ]}},
        "aggs": {
            "by_ip": {
                "terms": {"field": "data.srcip", "size": 50,
                          "min_doc_count": THRESHOLD},
                "aggs": {
                    "hosts": {"terms": {"field": "agent.name", "size": 10}},
                    "users": {"terms": {"field": "data.srcuser", "size": 10}},
                },
            }
        },
    }
    res = client.search(index="wazuh-alerts-*", body=body)
    return res["aggregations"]["by_ip"]["buckets"]

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

      import json, pathlib

STATE = pathlib.Path("//6ef4e6a1-9d49-47ac-bfed-170f67a815cf.selcdn.net/var/lib/wz-alerts/seen.json")
COOLDOWN = timedelta(hours=1)

def load_state():
    if STATE.exists():
        return {k: datetime.fromisoformat(v)
                for k, v in json.loads(STATE.read_text()).items()}
    return {}

def should_notify(seen, key, now):
    last = seen.get(key)
    if last and now - last < COOLDOWN:
        return False
    seen[key] = now
    return True

def save_state(seen):
    STATE.parent.mkdir(parents=True, exist_ok=True)
    STATE.write_text(json.dumps({k: v.isoformat() for k, v in seen.items()}))

Более «взрослый» вариант — писать состояние в свой же индекс в том же кластере, чтобы скрипт был stateless и мог ездить между хостами. Кстати, о своих индексах. 

Пишем обратно: свой индекс с обогащением

До этого мы только читали данные из Wazuh Indexer. Но OpenSearch можно использовать и как место хранения собственных результатов обработки. Классический кейс: обогатить алерты контекстом, которого в Wazuh нет и взяться ему неоткуда — например, владелец сервиса, критичность системы, номер тикета, вердикт аналитика.

      from opensearchpy import helpers

# один раз — маппинг, чтобы не полагаться на динамическое определение типов
client.indices.create(
    index="soc-enriched-000001",
    body={
        "aliases": {"soc-enriched": {"is_write_index": True}},
        "settings": {"number_of_shards": 1, "number_of_replicas": 1},
        "mappings": {
            "properties": {
                "@timestamp": {"type": "date"},
                "alert_id": {"type": "keyword"},
                "agent_name": {"type": "keyword"},
                "rule_id": {"type": "keyword"},
                "rule_level": {"type": "integer"},
                "business_unit": {"type": "keyword"},
                "criticality": {"type": "keyword"},
                "owner": {"type": "keyword"},
                "ticket": {"type": "keyword"},
                "verdict": {"type": "keyword"},
            }
        },
    },
    ignore=400,   # уже существует — не страшно
)

CMDB = {
    "web-prod-02": {"business_unit": "ecommerce", "criticality": "high",
                    "owner": "team-web"},
    "db-prod-01":  {"business_unit": "billing",   "criticality": "critical",
                    "owner": "team-db"},
}

def enrich(alerts):
    for hit in alerts:
        s = hit["_source"]
        meta = CMDB.get(s["agent"]["name"], {})
        yield {
            "_index": "soc-enriched",
            "_id": s["id"],          # id алерта как ключ — идемпотентно
            "_source": {
                "@timestamp": s["timestamp"],
                "alert_id": s["id"],
                "agent_name": s["agent"]["name"],
                "rule_id": s["rule"]["id"],
                "rule_level": s["rule"]["level"],
                **meta,
            },
        }

ok, errors = helpers.bulk(
    client,
    enrich(recent_alerts),
    chunk_size=1000,
    request_timeout=120,
    raise_on_error=False,
)
print(f"записано {ok}, ошибок {len(errors)}")

Разберем несколько важных деталей в этом фрагменте.

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

Алиас с is_write_index — задел на ротацию. В будущем вы просто создадите soc-enriched-000002 и переключите алиас, а код менять не придется. Заодно на алиас можно повесить ISM-политику через плагин управления индексами.

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

Дальше поверх soc-enriched строится обычный index pattern в дашборде Wazuh, и вот у вас уже дашборд «алерты по бизнес-юнитам», которого в поставке нет и не будет.

Сколько это весит и как не уронить кластер

Несколько правил, выведенных болезненным путем. Они помогут вам не создавать лишнюю нагрузку на индексер.

Всегда сужайте запрос по времени. Запрос без фильтра по timestamp уйдет в индексы за всю историю. На инсталляции с годовой ретенцией это могут быть десятки и сотни шардов на один запрос.

Бейте по конкретным индексам, а не по маске. Для ежедневных индексов Wazuh запрос к конкретным дням позволяет не опрашивать лишние шарды. Например, вместо wazuh-alerts-* для отчета за неделю лучше сформировать список нужных индексов:

      from datetime import date, timedelta

def daily_indices(start: date, end: date, prefix="wazuh-alerts-4.x-"):
    d, out = start, []
    while d <= end:
        out.append(f"{prefix}{d:%Y.%m.%d}")
        d += timedelta(days=1)
    return ",".join(out)

idx = daily_indices(date(2026, 7, 1), date(2026, 7, 31))

Если часть индексов могла отсутствовать, добавьте ignore_unavailable=True в параметры запроса.

Сначала считайте, потом выгружайте. client.count() стоит значительно дешевле полной выборки. Проверить, что запрос вернет несколько тысяч документов, а не 40 миллионов, проще до запуска тяжелой обработки.

Ставьте потолок на выполнение. Параметр timeout в теле запроса (не тот, что в клиенте) заставит индексер вернуть частичный результат вместо того, чтобы «молотить» бесконечно:

      body = {"timeout": "30s", "size": 0, "aggs": {...}}

Ходите на дата-ноды, а не на мастер. Если кластер состоит из нескольких нод, клиент лучше направлять на дата-ноды, которые выполняют поиск. Не стоит использовать мастер-узлы как точку входа для тяжелой аналитики.

Тяжелые выгрузки запускайте вне пиковых часов (по ночам). Учитывайте, что индексер в Wazuh одновременно занят приемом потока от Filebeat. Полная выгрузка месяца в девять утра понедельника — это индексационные лаги и очередь на сервере.

Что меняется в 5.0

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

• wazuh-alerts-* заменяется новой схемой событий wazuh-events-v5-*, разделенной по категориям: security, system-activity, network-activity, access-management, applications, cloud-services и другим. Для необработанного потока предусмотрен отдельный индекс wazuh-events-raw-v5.
• Появляется wazuh-findings-v5-* — слой результатов анализа Security Analytics, где события дополнительно обогащаются контекстом и метаданными правил.
• Индексы состояний переходят в формат wazuh-states-v5-*. В текущей реализации 5.0 туда также добавляются данные FIM и SCA, которые в 4.x не представлены отдельными индексами.
• Появляются отдельные индексы для метрик (wazuh-metrics-*) и threat intelligence (wazuh-threatintel-*).

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

      INDEX = {
    "alerts": os.getenv("WZ_IDX_ALERTS", "wazuh-alerts-*"),
    "vulns":  os.getenv("WZ_IDX_VULNS",  "wazuh-states-vulnerabilities-*"),
    "packages": os.getenv("WZ_IDX_PKGS", "wazuh-states-inventory-packages-*"),
}

То же касается полей документов: лучше держать отдельный слой соответствия между вашими внутренними названиями и реальной схемой OpenSearch.

Заключение

Wazuh — это OpenSearch с хорошей обвязкой, которая закрывает типовые сценарии. А все, что за их пределами, делается сотней строк на Python: свои отчеты, своя корреляция, свои пороги, свой алертинг, обогащение из CMDB, выгрузка в BI.

Если резюмировать до трех пунктов:

• считайте данные на стороне индексера с помощью агрегаций. Тянуть «сырье» в pandas ради groupby — это медленно;
• Используйте отдельного read-only пользователя, а не admin.
• Выносите имена индексов в конфигурацию. Версия 5.0 рано или поздно придет, и придет с новыми именами.