Модуль security-events-manager: FAQ

Нужен ли мне этот модуль?

Имеет смысл использовать его, если вы хотите:

• Собирать события безопасности (audit, runtime, приложение) в едином каноническом формате.
• Маршрутизировать события в Loki, Elasticsearch, Splunk и др. по источнику и Severity.
• Дать другим командам описывать извлечение событий (Shipper), а сами владеть реестром (SecurityEventDefinition) и приёмниками.

Если нужна только общая доставка логов без валидации и маршрутизации событий, достаточно модуля log-shipper.

Как включить доставку в Loki?

Доставка в Loki, расположенный в кластере, включена по умолчанию.

Для отправки во внешний Loki требуется выполнить следующие действия:

1. Создайте ClusterSecurityEventDestination с spec.type: Loki и укажите эндпоинт, аутентификацию и TLS для Loki. Подробнее о конфигурации и примерах — в CR и Examples.
2. Создайте ClusterSecurityEventConfig: в spec.destinations укажите этот приёмник и задайте разрешённые источники в spec.enabledSources или spec.enabledSourcesMasks.
3. Убедитесь, что есть минимум один SecurityEventDefinition (контроллер создаст ClusterLogDestination для шлюза) и хотя бы один Shipper, производящий события из источников, разрешённых в ClusterSecurityEventConfig.

События не доходят до приёмника. Что проверить?

• Список источников — в ClusterSecurityEventConfig источник должен быть указан в enabledSources или enabledSourcesMasks. Формат источника —
  • clusterSecurityEventShipper/<имя ClusterSecurityEventShipper>/<source> (например clusterSecurityEventShipper/kube-audit/kube-apiserver)
  • podSecurityEventShipper/<namespace>/<имя PodSecurityEventShipper>/<source> (например podSecurityEventShipper/my-ns/audit/my-component)
• Маски — при использовании enabledSourcesMasks символ * сопоставляется с любой подстрокой, включая /. Нельзя задавать одновременно enabledSources и enabledSourcesMasks.
• Severity — параметр defaultSeverityThreshold фильтрует по .event.severity; события ниже порога, заданного в ClusterSecurityEventConfig, не отправляются в соответствующие приёмники.
• Реестр — при использовании SecurityEventDefinition события, пара которых (event.code, source.component) не найдена в реестре, отбрасываются на шлюзе. Убедитесь, что produces[].eventCode и source в Shipper совпадают с SecurityEventDefinition.

Подробнее про проверку конфигурационного файла и логов шлюза можно прочитать в расширенном использовании.

В чём разница между PodSecurityEventShipper и ClusterSecurityEventShipper?

• PodSecurityEventShipper — оперирует объектами в рамках одного неймспейса. Читает логи подов в этом неймспейсе. Используйте для событий приложений или привязанных к неймспейсу.
• ClusterSecurityEventShipper — cluster-scoped. Может читать логи подов в выбранных неймспейсах или файлы на узлах. Используйте для сбора событий с однотипных подов, расположенных в нескольких неймспейсах (например, поды dex-authenticator), или сбора логов на узлах.

Оба производят события с полями eventCode и source; маршрутизация в ClusterSecurityEventConfig использует один и тот же формат источника с префиксом podSecurityEventShipper/... или clusterSecurityEventShipper/....

Как добавить новый тип события?

1. Создайте или используйте существующий SecurityEventDefinition с нужными code, severity, category, source и fields (обязательные по необходимости).
2. Создайте или обновите PodSecurityEventShipper или ClusterSecurityEventShipper с элементом spec, в котором produces[].eventCode совпадает с этим definition, а source — с source из definition. Настройте input, parser или parserRef, при необходимости produces[].transform и enrich.
3. Убедитесь, что ClusterSecurityEventConfig включает этот источник (enabledSources или enabledSourcesMasks) и ссылается на приёмник, куда должны попадать события.

severity и category задаются только в SecurityEventDefinition, а не в Shipper.

Как отправлять события во внешнюю систему (OpenSearch, Elasticsearch, Splunk и др.)?

Модуль security-events-manager поддерживает отправку событий в несколько типов внешних систем хранения и аналитики: Loki, Elasticsearch, Kafka, Splunk (HEC), Vector, File, Socket и Console.

Рассмотрим подключение на примере сервиса OpenSearch.

OpenSearch обратно совместим с Elasticsearch API, поэтому для отправки событий используется тип назначения Elasticsearch и эндпоинты _bulk/_search.

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

1. Настроить приёмник событий — ресурс ClusterSecurityEventDestination.
2. Настроить правила отправки — ресурс ClusterSecurityEventConfig.
3. Убедиться, что в кластере есть хотя бы один ClusterSecurityEventShipper или PodSecurityEventShipper, производящий события из разрешённых источников.

Настройка приёмника

Приёмник настраивается через ресурс ClusterSecurityEventDestination.

Для OpenSearch в поле endpoint указывается HTTPS-адрес API (через Ingress — https://opensearch-api.example.com, или внутренний — https://opensearch-cluster-master:9200). Стратегии авторизации — None, Bearer или Basic. Проверка сертификата сервера настраивается в блоке tls. Поле index задаёт имя индекса или шаблон индексации, например security-events-%Y.%m.%d для ежедневной ротации индекса.

Пример:

apiVersion: security.deckhouse.io/v1alpha1
kind: ClusterSecurityEventDestination
metadata:
  name: opensearch-external
spec:
  type: Elasticsearch
  elasticsearch:
    endpoint: https://opensearch-api.example.com
    # Ежедневная ротация индекса. Уберите суффикс для единого индекса.
    index: security-events-%Y.%m.%d
    auth:
      # None | Bearer | Basic.
      strategy: Basic
      username: admin
      password: P@ssw0rd-CHANGE-ME
    tls:
      # Для публичного сертификата Let's Encrypt через Ingress.
      # Если у шлюза нет системного CA, проверку можно отключить.
      # Для production-окружения задайте ca (Base64 PEM) и включите проверку.
      verifyCertificate: false
      verifyHostname: false
      # ca: LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0t...

Для production-окружения рекомендуется завести отдельного пользователя в OpenSearch с правами только на запись в целевой индекс.

Вместо хранения учётных данных в спецификации CR их можно подключать из Secret через passwordSecretRef/tokenSecretRef. Secret должен находиться в namespace d8-security-events-manager с ключом value в поле data и меткой security-events-manager.deckhouse.io/credential-secret: "true". Inline-поля учётных данных и их *SecretRef-аналоги взаимно исключают друг друга.

Настройка правил отправки

Правила отправки настраиваются через ресурс ClusterSecurityEventConfig.

В поле destinations указывается имя приёмника, заданное в metadata.name ресурса ClusterSecurityEventDestination. В поле enabledSources (или enabledSourcesMasks) перечисляются источники, события из которых отправляются в приёмник. Поле defaultSeverityThreshold задаёт минимальный уровень серьёзности — события с уровнем ниже указанного отбрасываются.

Пример:

apiVersion: security.deckhouse.io/v1alpha1
kind: ClusterSecurityEventConfig
metadata:
  name: opensearch-routing
spec:
  # Low | Medium | High — отбрасывает события ниже порога.
  defaultSeverityThreshold: Low
  enabledSourcesMasks:
    - clusterSecurityEventShipper/* # Все кластерные источники.
    - podSecurityEventShipper/* # Все источники на уровне неймспейса.
  destinations:
    - opensearch-external # Имя ресурса ClusterSecurityEventDestination.

Можно создать несколько ресурсов ClusterSecurityEventDestination и указать их в destinations одного ClusterSecurityEventConfig — отобранные события будут отправлены во все приёмники одновременно. Для разбиения потока по уровню серьёзности используются несколько ресурсов ClusterSecurityEventConfig с разными значениями defaultSeverityThreshold и непересекающимися destinations или enabledSources.

Как собирать события sshd (успешные и неуспешные SSH-подключения)?

Рассмотрим сбор событий на примере Ubuntu и файла /var/log/auth.log. Источник типа File читает файлы на узлах через DaemonSet log-shipper, поэтому путь должен быть доступен для чтения на каждой ноде.

Пример строк лога:

2026-07-28T05:41:36.390129+00:00 master-0 sshd[1127780]: Accepted publickey for a.dyakonov from 10.12.0.1 port 60280 ssh2: RSA SHA256:...
2026-07-28T05:53:10.932848+00:00 master-0 sshd[1194201]: Invalid user test from 127.0.0.1 port 35176

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

1. SecurityEventDefinition — описания событий (успешное и неуспешное подключение).
2. ClusterSecurityEventShipper — пайплайн: file-источник, regex-парсер syslog-строки, правила produces для детекции.

После создания ресурсов не забудьте включить источник clusterSecurityEventShipper/sshd/sshd в ClusterSecurityEventConfig (enabledSources или enabledSourcesMasks), иначе события будут отброшены на шлюзе.

Шаг 1. Определения событий

severity и category задаются только в SecurityEventDefinition. Поле source должно совпадать со значением source в Shipper.

apiVersion: security.deckhouse.io/v1alpha1
kind: SecurityEventDefinition
metadata:
  name: ssh-login-success
spec:
  code: SSH_LOGIN_SUCCESS
  category: Auth
  severity: Low
  description: "Successful SSH login"
  descriptionRu: "Успешное SSH-подключение"
  # metadata.docs.desc — описание для отображения дополнительной информации в Deckhouse Console UI.
  metadata:
    docs:
      desc:
        ru: |
          Обнаружено успешное SSH-подключение к узлу. Событие фиксирует аутентификацию пользователя по открытому ключу или паролю.
        en: |
          Detected a successful SSH login to the node. The event records user authentication via public key or password.
  source: sshd
  fields:
    - name: actor.id
      required: true
    - name: metadata.extra.src_ip
    - name: metadata.extra.src_port
    - name: metadata.extra.auth_method
---
apiVersion: security.deckhouse.io/v1alpha1
kind: SecurityEventDefinition
metadata:
  name: ssh-login-failed
spec:
  code: SSH_LOGIN_FAILED
  category: Auth
  severity: Medium
  description: "Failed SSH login attempt"
  descriptionRu: "Неуспешная попытка SSH-подключения"
  # metadata.docs.desc — описание для отображения дополнительной информации в Deckhouse Console UI.
  metadata:
    docs:
      desc:
        ru: |
          Обнаружена неуспешная попытка SSH-подключения к узлу. Событие фиксирует вход с неверным пользователем, неверным ключом или отклонённой аутентификацией.
        en: |
          Detected a failed SSH login attempt to the node. The event records a login with an invalid user, wrong key, or rejected authentication.
  source: sshd
  fields:
    - name: actor.id
      required: true
    - name: metadata.extra.src_ip
    - name: metadata.extra.src_port

Шаг 2. ClusterSecurityEventShipper

Парсер использует тип Regex с несколькими шаблонами: первый успешно совпавший шаблон выигрывает, именованные группы записываются в .parsed_data. Правила extract (поле message) фильтруют строки на стороне log-shipper — в шлюз попадают только строки, соответствующие sshd-событиям. Правила transform в producesDefaults переносят распарсенные поля в каноническую модель SecurityEvent, а enrich устанавливает source.component.

apiVersion: security.deckhouse.io/v1alpha1
kind: ClusterSecurityEventShipper
metadata:
  name: sshd
spec:
  - source: sshd
    input:
      type: File
      files:
        - /var/log/auth.log
    parser:
      - name: file
        parser:
          type: Regex
          regex:
            patterns:
              # Accepted <method> for <user> from <ip> port <port> ssh2
              - '^(?P<syslog_timestamp>\S+)\s+(?P<hostname>\S+)\s+sshd\[(?P<pid>\d+)\]:\s+Accepted\s+(?P<auth_method>\S+)\s+for\s+(?P<user>\S+)\s+from\s+(?P<src_ip>\S+)\s+port\s+(?P<src_port>\d+).*$'
              # Invalid user <user> from <ip> port <port>
              - '^(?P<syslog_timestamp>\S+)\s+(?P<hostname>\S+)\s+sshd\[(?P<pid>\d+)\]:\s+Invalid\s+user\s+(?P<user>\S+)\s+from\s+(?P<src_ip>\S+)\s+port\s+(?P<src_port>\d+).*$'
              # Failed <method> for [invalid user ]<user> from <ip> port <port>
              - '^(?P<syslog_timestamp>\S+)\s+(?P<hostname>\S+)\s+sshd\[(?P<pid>\d+)\]:\s+Failed\s+(?P<auth_method>\S+)\s+for\s+(?:invalid\s+user\s+)?(?P<user>\S+)\s+from\s+(?P<src_ip>\S+)\s+port\s+(?P<src_port>\d+).*$'
        fields:
          - name: src_port
            type: Int
          - name: pid
            type: Int
    producesDefaults:
      transform:
        - key: timestamp
          value: syslog_timestamp
        - key: actor.id
          value: user
        - key: metadata.extra.src_ip
          value: src_ip
        - key: metadata.extra.src_port
          value: src_port
        - key: metadata.extra.hostname
          value: hostname
        - key: metadata.extra.auth_method
          value: auth_method
        - key: metadata.extra.pid
          value: pid
      enrich:
        - target: source.component
          source: Static
          value: sshd
    produces:
      - eventCode: SSH_LOGIN_SUCCESS
        extract:
          field: message
          operator: Regex
          values:
            - '.*sshd\[.*Accepted\s+\S+\s+for\s+\S+\s+from\s+\S+\s+port\s+\d+.*'
      - eventCode: SSH_LOGIN_FAILED
        extract:
          field: message
          operator: Regex
          values:
            - '.*sshd\[.*Invalid\s+user\s+\S+\s+from\s+\S+\s+port\s+\d+.*'
            - '.*sshd\[.*Failed\s+\S+\s+for\s+(?:invalid\s+user\s+)?\S+\s+from\s+\S+\s+port\s+\d+.*'

Конфигурация буфера

Что происходит, когда назначение недоступно?

Когда назначение (например, Loki, Elasticsearch) временно недоступно, gateway буферизует события. Поведение зависит от настроек буфера:

• При whenFull: Block (по умолчанию): gateway применяет обратное давление — пайплайн замедляется, но НЕТ потерь событий. Агенты log-shipper также буферизуют на своей стороне, каскадно передавая обратное давление к источникам.
• При whenFull: DropNewest: новые события отбрасываются при переполнении буфера. Происходит потеря данных.

Алерт D8SecurityEventsManagerGatewayBufferHighUsage срабатывает при использовании буфера > 80% в течение 10 минут, а D8SecurityEventsManagerGatewayBufferEventsDropped — при отбрасывании событий.

Почему gateway работает медленно?

Если gateway кажется медленным, возможно, он применяет обратное давление из-за переполнения буфера (whenFull: Block). Проверьте:

1. Доступны ли назначения (проверьте kubectl get clustersecurityeventdestination -o wide и состояние эндпоинтов).
2. Метрики использования буфера на дашборде Grafana.
3. Логи gateway на ошибки sink: kubectl -n d8-security-events-manager logs deploy/gateway -c vector --tail=500 | grep -E "buffer|backpressure|sink|error"

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

Сколько диска использует буфер?

При типе буфера Disk (по умолчанию):

• Sink gateway: до gateway.buffer.maxSize (по умолчанию 512Mi) на реплику.
• Агенты log-shipper: до gateway.logShipperBuffer.maxSize (по умолчанию 257Mi) на узел.
• emptyDir vector-data шлюза имеет sizeLimit, соответствующий maxSize, при использовании дискового буфера.

Общее использование диска gateway = gateway.replicas × gateway.buffer.maxSize.

Block или DropNewest — что использовать?

• Block (по умолчанию): используйте в production. Пайплайн замедляется при обратном давлении, но события не теряются. Правильный выбор для событий безопасности.
• DropNewest: используйте только в тестовых/dev-окружениях, где потеря данных допустима и вы не хотите, чтобы обратное давление влияло на агентов log-shipper.

Как рассчитать размер буфера?

Длительность буфера (часы) = maxSize_байт / (средний_размер_события_байт × событий_в_секунду × 3600)

При значениях по умолчанию (512Mi = 536870912 байт, 500 байт/событие, 10 соб/с): ~30 часов буферизации. При высокой нагрузке (10 000 соб/с, 1 КБ события): ~9 минут — увеличьте maxSize соответствующим образом.

Подробнее см. Расширенное использование — Настройка буфера.