Перейти к основному содержимому
Версия: 2.5

Передача событий аудита

События аудита Памир передаются через RabbitMQ и могут быть обработаны встроенным в систему Logstash.

Описание формирования событий аудита и перечень ключей маршрутизации приведены в документе Действия в системе.

Принцип работы​

RabbitMQ принимает сообщения Памир и публикует их в exchange, например pamir_audit_exchange.

Logstash подключается к RabbitMQ, создает или использует очереди для аудита, извлекает нужные события по ключу маршрутизации, фильтрует их по роли пользователя и по названию прикладного объекта, после чего:

  • передает события во встроенный OpenSearch;
  • при необходимости передает события во внешние системы.

Схема передачи событий:

Использование Logstash​

Logstash входит в состав дистрибутива Памир и используется для получения событий из RabbitMQ, их фильтрации и последующей передачи:

  • во встроенный OpenSearch;
  • во внешние системы;
  • в локальные файлы.

Для добавления пользовательских pipeline можно использовать один из следующих вариантов:

  1. запустить отдельный экземпляр Logstash для пользовательских pipeline — рекомендуемый вариант;
  2. подключить пользовательский pipeline к системному экземпляру Logstash, заменив его файл pipelines.yml — нерекомендуемый вариант.

Рекомендуемый вариант: отдельный сервис Logstash​

Рекомендуется запускать пользовательские pipeline в отдельном сервисе logstash-user.

Для передачи событий во внешние системы также следует использовать пользовательский сервис logstash-user. Не добавляйте пользовательские блоки output в конфигурацию системного сервиса logstash.

При таком подходе:

  • системный сервис logstash продолжает использовать встроенный файл pipelines.yml;
  • пользовательский сервис использует отдельный файл pipelines.user.yml;
  • обновления состава системных pipeline не требуют изменения пользовательской конфигурации;
  • пользовательский сервис наследует базовые настройки сервиса logstash из docker-compose.deps.yml, а необходимые параметры переопределяются в docker-compose.additional.yml.

Создайте файл:

/home/pamir/.pamir/data/logstash/pipelines.user.yml

Пример содержимого файла:

- pipeline.id: mes-test-case
  path.config: "/usr/share/logstash/pipeline/mes-test-case.conf"

Каждый дополнительный pipeline необходимо добавить в этот файл отдельным элементом:

- pipeline.id: mes-test-case
  path.config: "/usr/share/logstash/pipeline/mes-test-case.conf"

- pipeline.id: external-audit
  path.config: "/usr/share/logstash/pipeline/external-audit.conf"

Значение pipeline.id должно быть уникальным в пределах одного экземпляра Logstash.

Конфигурацию pipeline необходимо сохранить в отдельном файле, например:

/home/pamir/.pamir/data/logstash/pipeline/mes-test-case.conf

В файл docker-compose.additional.yml добавьте отдельный сервис logstash-user:

Сервис logstash-user наследует настройки системного сервиса logstash из файла docker-compose.deps.yml. Параметры, указанные в logstash-user, переопределяют или дополняют унаследованную конфигурацию.

services:
  logstash-user:
    extends:
      file: docker-compose.deps.yml
      service: logstash

    ports:
      - "172.20.4.121:38080:38080"

    volumes:
      - type: volume
        source: logstash-user-tmp
        target: /tmp

      - type: volume
        source: logstash-user-config
        target: /usr/share/logstash/config

      - type: volume
        source: logstash-user-data
        target: /usr/share/logstash/data

      - type: bind
        source: /home/pamir/.pamir/data/logstash/pipelines.user.yml
        target: /usr/share/logstash/config/pipelines.yml
        read_only: true

      - type: bind
        source: /home/pamir/.pamir/data/logstash/pipeline/mes-test-case.conf
        target: /usr/share/logstash/pipeline/mes-test-case.conf
        read_only: true

volumes:
  logstash-user-tmp:
  logstash-user-config:
  logstash-user-data:

Если разделы services и volumes уже присутствуют в docker-compose.additional.yml, необходимо добавить в них только сервис logstash-user и соответствующие тома. Повторно объявлять разделы верхнего уровня не требуется.

Для каждого дополнительного файла конфигурации pipeline необходимо добавить отдельный bind mount:

      - type: bind
        source: /home/pamir/.pamir/data/logstash/pipeline/external-audit.conf
        target: /usr/share/logstash/pipeline/external-audit.conf
        read_only: true

Нерекомендуемый вариант: изменение системного сервиса Logstash​

Пользовательский pipeline можно подключить непосредственно к системному сервису logstash.

Для этого необходимо смонтировать:

  • пользовательский файл конфигурации pipeline;
  • собственный файл pipelines.yml поверх системного файла из образа.

При bind mount файла:

/usr/share/logstash/config/pipelines.yml

встроенный файл pipelines.yml полностью скрывается и заменяется файлом с узла Docker.

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

Просмотр текущей конфигурации:

docker exec -it pamir-logstash-1 \
  cat /usr/share/logstash/config/pipelines.yml

Сохранение текущей конфигурации непосредственно в файл на узле Docker:

docker exec pamir-logstash-1 \
  cat /usr/share/logstash/config/pipelines.yml \
  > /home/pamir/.pamir/data/logstash/pipelines.yml

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

Пример файла:

/home/pamir/.pamir/data/logstash/pipelines.yml

Пример содержимого:

- pipeline.id: pamir_audit
  path.config: "/usr/share/logstash/pipeline/pamir-audit.conf"

- pipeline.id: pamir-user-activity
  path.config: "/usr/share/logstash/pipeline/pamir-user-activity.conf"

- pipeline.id: pamir-enrichment-statistic
  path.config: "/usr/share/logstash/pipeline/pamir-enrichment-statistic.conf"

- pipeline.id: pamir-sec-ext-service-access
  path.config: "/usr/share/logstash/pipeline/pamir-sec-ext-service-access.conf"

- pipeline.id: pamir-import-statistic
  path.config: "/usr/share/logstash/pipeline/pamir-import-statistic.conf"

- pipeline.id: mes-test-case
  path.config: "/usr/share/logstash/pipeline/mes-test-case.conf"

После этого в файл docker-compose.additional.yml добавьте монтирование пользовательского pipeline и файла pipelines.yml:

services:
  logstash:
    volumes:
      - /home/pamir/.pamir/data/logstash/pipeline/mes-test-case.conf:/usr/share/logstash/pipeline/mes-test-case.conf:ro
      - /home/pamir/.pamir/data/logstash/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro

Этот вариант не рекомендуется, поскольку файл pipelines.yml входит в состав системного образа и может изменяться при обновлении Памир.

После обновления могут быть:

  • добавлены новые системные pipeline;
  • удалены устаревшие системные pipeline;
  • изменены идентификаторы pipeline;
  • изменены пути к файлам конфигурации.

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

После каждого обновления необходимо:

  • Получить актуальный системный файл;
  • Перенести изменения системных pipeline в файл;
  • Повторно добавить в него пользовательские записи.

По этой причине для пользовательских pipeline следует использовать отдельный сервис logstash-user.

Настройка pipeline Logstash​

Параметры подключения для pipeline Logstash задаются в файле окружения, подключённом через env_file.

Пример переменных окружения:

RABBITMQ_HOST=rabbitmq
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_VHOST=vhost

OPENSEARCH_LOGSTASH_PASSWORD=pswd

Конфигурацию pipeline необходимо записать в отдельный файл, например pamir-audit.conf, и подключить его через volumes и pipelines.yml.

Если пользовательский pipeline должен получать системные события параллельно с системным pipeline, задайте для него отдельную очередь RabbitMQ. Имя queue не должно совпадать с именем очереди системного pipeline: несколько потребителей одной очереди распределяют сообщения между собой, а не получают отдельные копии каждого сообщения. Значения exchange и key выберите в соответствии с передаваемым типом событий. Если внешний блок output добавляется в уже существующий pipeline, отдельная очередь не требуется.

Пример конфигурации:

input {
  rabbitmq {
    host => "${RABBITMQ_HOST:rabbitmq}"
    port => "${RABBITMQ_PORT:5672}"
    user => "${RABBITMQ_USER}"
    password => "${RABBITMQ_PASSWORD}"
    vhost => "${RABBITMQ_VHOST:vhost}"

    # Системные события приложения в exchange RabbitMQ типа topic
    exchange => "pamir_audit_exchange"
    exchange_type => "topic"
    durable => true

    # Routing key для фильтрации сообщений.
    # Перечень событий и ключей маршрутизации приведён в разделе Действия в системе.
    key => "#"

    # Для пользовательского pipeline используется отдельная очередь.
    # Logstash создаёт её автоматически, если она отсутствует.
    queue => "logstash-user-audit-external-queue"

    auto_delete => false
    exclusive => false
    durable => true

    ack => true
    prefetch_count => 200
    passive => false

    metadata_enabled => "none"

    codec => json { ecs_compatibility => "disabled" }
  }
}

filter {
  date {
    match => ["timestamp", "ISO8601"]
    target => "@timestamp"
  }

  mutate {
    remove_field => ["timestamp"]
  }

  # Разделение событий по ролям пользователей.
  if [user_role] in ["pamir_root", "pamir_admin", "pamir_admin_ib"] {
    mutate {
      add_field => { "[@metadata][index_prefix]" => "pamir-sec-crud-admin-logs" }
    }
  } else {
    mutate {
      add_field => { "[@metadata][index_prefix]" => "pamir-sec-crud-user-logs" }
    }
  }
}

output {
  # --- OpenSearch ---
  # Передача событий аудита в OpenSearch для централизованного хранения, поиска и последующего анализа.
  opensearch {
    hosts => ["https://opensearch:9200"]
    auth_type => {
      type => "basic"
      user => "pamir-logstash-user"
      password => "${OPENSEARCH_LOGSTASH_PASSWORD}"
    }
    ssl_certificate_verification => false
    index => "%{[@metadata][index_prefix]}-%{+YYYY.MM.dd}"
    manage_template => false
  }

  # --- SYSLOG ---
  # Передача событий аудита на внешний сервер сбора и анализа
  syslog {
    id => "audit_syslog_tls"
    host => "${SYSLOG_HOST:syslog.example.local}"
    port => "${SYSLOG_PORT:6514}"
    protocol => "ssl-tcp"
    rfc => "rfc5424"
    appname => "pamir-audit"
    facility => "local0"
    severity => "informational"
    sourcehost => "%{[host]}"
    message => '{"event_type":"%{[event_type]}","user_id":"%{[user_id]}","user_login":"%{[user_login]}","user_role":"%{[user_role]}","endpoint":"%{[endpoint]}","parameters":"%{[parameters]}","response_status_code":"%{[response_status_code]}","@timestamp":"%{[@timestamp]}"}'
    ssl_certificate_authorities => ["/usr/share/logstash/config/certs/syslog-ca.crt"]
    ssl_verify => true
  }

  # --- REST API ---
  # Передача событий аудита во внешний сервис по REST API с использованием HTTPS.
  http {
    id => "audit_rest_https"
    url => "https://audit-collector.example.local/api/v1/audit"
    http_method => "post"
    format => "json"
    content_type => "application/json"
    headers => {
      "Authorization" => "Bearer ${AUDIT_API_TOKEN}"
      "X-Source-System" => "pamir"
    }
    automatic_retries => 2
    retry_failed => true
    ssl_enabled => true
    ssl_verification_mode => "full"
    ssl_certificate_authorities => ["/usr/share/logstash/config/certs/api-ca.crt"]
  }

  # --- JDBC ---
  # Передача событий аудита во внешнюю реляционную базу данных по JDBC.

  jdbc {
    driver_class => "org.postgresql.Driver"
    connection_string => "jdbc:postgresql://postgres:5432/auditdb?ssl=true&sslmode=verify-full"
    username => "${JDBC_USER}"
    password => "${JDBC_PASSWORD}"
    statement => [
      "INSERT INTO audit_events (event_time, event_type, user_id, user_login, user_role, endpoint, parameters, response_status_code, payload) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
      "@timestamp",
      "event_type",
      "user_id",
      "user_login",
      "user_role",
      "endpoint",
      "parameters",
      "response_status_code",
      "payload"
    ]
  }
}

В приведённой конфигурации:

  • блок input.rabbitmq подключает Logstash к RabbitMQ и получает события из exchange pamir_audit_exchange через очередь logstash-audit-queue;
  • параметр key задаёт перечень ключей маршрутизации, по которым будут отбираться события аудита;
  • блок filter.date преобразует время события в поле @timestamp;
  • блок filter.mutate удаляет исходное поле timestamp после преобразования;
  • блок filter с проверкой user_role разделяет события административных пользователей и остальных пользователей по разным индексам OpenSearch;
  • блок output.opensearch передаёт события во встроенный OpenSearch;
  • дополнительные блоки output могут использоваться для передачи событий во внешние системы.

Таким образом конфигурация позволяет получать события аудита из RabbitMQ, фильтровать их по ключу маршрутизации, роли пользователя и названию прикладного объекта, а затем передавать их в OpenSearch или во внешние системы.

Настройка перечня объектов доступа​

Настройка перечня объектов доступа, для которых регистрируются попытки обращения, выполняется на двух уровнях:

  1. на уровне подписки на события RabbitMQ по routing key;
  2. на уровне фильтрации событий в Logstash.

Таким образом обеспечивается возможность настройки перечня объектов доступа:

  • по сервису;
  • по классу прикладного объекта;
  • по роли пользователя;
  • по имени конкретного прикладного объекта;
  • по параметрам пути запроса.

Фильтрация по ключу маршрутизации​

Фильтрация по routing key задаёт перечень событий, которые Logstash получает из RabbitMQ.

Например:

  • # - все события;
  • srm.# - все события сервиса СРМ;
  • srm.user_action.dashboard.* - все события, связанные с дашбордами;
  • srm.user_action.ci_type.* - все события, связанные с типами CI;
  • srm.user_action.ci_type.CITypeCreated - конкретное событие, создание типа CI.

Таким образом на уровне обращения к RabbitMQ можно задать перечень прикладных объектов, для которых будут регистрироваться попытки обращения.

Фильтрация по роли пользователя​

Фильтрация по ролям выполняется в блоке filter конфигурации Logstash.

Пример:

if [user_role] in ["pamir_root", "pamir_admin", "pamir_admin_ib"] {
  mutate {
    add_field => { "[@metadata][index_prefix]" => "pamir-sec-crud-admin-logs" }
  }
} else {
  mutate {
    add_field => { "[@metadata][index_prefix]" => "pamir-sec-crud-user-logs" }
  }
}

В этом примере:

  • события пользователей с ролями pamir_root, pamir_admin, pamir_admin_ib помещаются в индекс административных действий;
  • события остальных пользователей помещаются в индекс пользовательских действий.

При необходимости можно регистрировать события только для отдельных ролей:

if [user_role] not in ["pamir_root", "pamir_admin", "pamir_admin_ib"] {
  drop {}
}

Фильтрация по названию прикладного объекта​

Фильтрация по названию прикладного объекта выполняется после разбора поля payload.

Для событий:

  • srm.user_action.ci_type.* имя объекта, как правило, находится в payload_json.name;
  • srm.user_action.dashboard.* имя объекта может находиться в payload_json.title или payload_json.name.

Пример разбора поля payload:

json {
  source => "payload"
  target => "payload_json"
  skip_on_invalid_json => true
}

После этого можно использовать фильтрацию по названию объекта.

Пример фильтрации по названию типа CI:

if [event_type] =~ /^srm\.user_action\.ci_type\./ and [payload_json][name] != "audit_log" {
  drop {}
}

Пример фильтрации по названию дашборда:

if [event_type] =~ /^srm\.user_action\.dashboard\./ and [payload_json][title] != "audit" {
  drop {}
}

Таким образом можно регистрировать попытки обращения только к прикладным объектам с заданным именем.

Фильтрация по параметрам пути запроса​

Фильтрация по параметрам пути запроса может использоваться в случаях, когда имя или идентификатор прикладного объекта передаётся в path_parameters.

Состав полей path_parameters зависит от конкретного endpoint.

Такой способ фильтрации может использоваться, например, при регистрации событий изменения дашбордов.

Пример события:

  • event_type: srm.user_action.dashboard.DashboardUpdated;
  • endpoint: /v4/dashboard/{dashboard_name};
  • path_parameters.dashboard_name: audit.

Пример фильтрации по имени дашборда:

if [event_type] =~ /^srm\.user_action\.dashboard\./ and [path_parameters][dashboard_name] != "audit" {
  drop {}
}

Таким образом можно регистрировать попытки обращения только к прикладным объектам, параметры которых передаются в пути запроса.

Передача событий во встроенный OpenSearch​

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

В приведённой конфигурации Logstash:

  • административные события записываются в индексы вида pamir-sec-crud-admin-logs-YYYY.MM.dd;
  • пользовательские события записываются в индексы вида pamir-sec-crud-user-logs-YYYY.MM.dd.

Разделение по индексам выполняется на основании поля user_role.

Такой подход позволяет:

  • отделить административные действия от пользовательских;
  • упростить поиск событий;
  • упростить настройку хранения данных.

Передача событий во внешние системы​

Для передачи событий аудита во внешние системы рекомендуется использовать пользовательский pipeline, запущенный в отдельном сервисе logstash-user. Конфигурацию системного сервиса logstash изменять не следует.

Во входном блоке rabbitmq отдельного пользовательского pipeline необходимо указать уникальную queue. Это особенно важно при передаче системных событий: если системный и пользовательский pipeline используют одну очередь, RabbitMQ распределяет события между ними и пользовательский pipeline не получает копию каждого события. Если внешний блок output добавляется в уже существующий pipeline, его input и имя очереди изменять не требуется.

Для включения интеграции необходимо:

  1. Создать пользовательский pipeline и подключить его к сервису logstash-user;
  2. настроить входной блок rabbitmq с отдельной очередью;
  3. добавить нужный блок output;
  4. задать параметры подключения в файле окружения, подключённом через env_file;
  5. при необходимости добавить сертификаты и иные файлы конфигурации.
подсказка

В зависимости от выбранного блока output может потребоваться установка соответствующего плагина Logstash, например logstash-output-syslog для блока syslog.

Установка пользовательских плагинов описана в Плагины Logstash.

Передача по Syslog​

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

В примере конфигурации используется защищённый вариант TLS/TCP.

Передача по REST API​

Передача событий аудита во внешний сервис может быть выполнена по REST API.

В примере конфигурации используется протокол HTTPS с проверкой сертификата сервера.

Передача по JDBC​

Передача событий аудита во внешнюю базу данных может быть выполнена по JDBC.

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

Передача событий в локальный файл​

Для сохранения журналов ИБ на локальный диск в Logstash может использоваться плагин file.

Подключение каталога для журналов​

Для записи журналов в локальные файлы необходимо добавить в docker-compose.additional.yml отдельный volume с каталогом для хранения логов.

Пример:

  logstash:
    volumes:
      - /home/pamir/.pamir/data/logstash/logs:/var/log/logstash

В приведённом примере:

  • каталог /home/pamir/.pamir/data/logstash/logs расположен на узле с Docker;
  • каталог /var/log/logstash доступен внутри контейнера logstash;
  • файлы журналов записываются в смонтированный каталог на узле.

Для отдельных видов журналов рекомендуется использовать отдельные подкаталоги, например pamir-sec-user-activity.

Настройка output​

Для записи событий в локальный файл необходимо добавить в pipeline блок output.

Пример:

output {
  file {
    path => "/var/log/logstash/pamir-sec-user-activity/pamir-sec-user-activity-%{+YYYY-MM-dd}.log"
    codec => json_lines
    flush_interval => 1
    create_if_deleted => true
  }
}

В приведённой конфигурации:

  • path задаёт путь к файлу журналов;
  • %{+YYYY-MM-dd} в имени файла обеспечивает разбиение файлов по дате события;
  • codec => json_lines задаёт запись каждого события отдельной строкой в формате JSON;
  • flush_interval => 1 задаёт интервал сброса буфера на диск;
  • create_if_deleted => true указывает, что при удалении файла он будет создан повторно при поступлении следующего события.

Результат работы​

При такой настройке Logstash записывает события в локальные файлы вида:

  • pamir-sec-user-activity-YYYY-MM-dd.log.

Таким образом обеспечивается разбиение журналов по дням.

Особенности использования​

Данная настройка обеспечивает разбиение файлов по дате, но не включает полноценную ротацию и политику хранения файлов средствами операционной системы.

Это означает, что:

  • старые файлы не удаляются автоматически;
  • сжатие файлов не выполняется автоматически;
  • ротация по размеру файла не выполняется автоматически.

При необходимости удаление старых файлов, сжатие и ротация по размеру должны быть дополнительно настроены средствами операционной системы.

Альтернативная передача событий с помощью vector​

Для передачи событий аудита также может использоваться утилита vector.

Подключение утилиты vector​

Для подключения утилиты vector в виде сервиса, необходимо в файл docker-compose.additional.yml добавить следующие строки:

  vector: 
    image: timberio/vector:latest-alpine
    networks: [pamir-net]
    profiles: [all, additional]
    restart: unless-stopped
    env_file:
      - .env
    volumes:
      - type: bind
        source: ./data/vector
        target: /etc/vector
    command: ["--config", "/etc/vector/vector.toml"]

Настройка утилиты vector​

Конфигурацию сервиса vector нужно записать в файл vector.toml. Путь к файлу vector.toml необходимо указать в файле docker-compose.additional.yml, в volumes.

Пример конфигурации:

[sources.rabbitmq]
type = "amqp"
connection_string = "amqp://${RABBITMQ_USER}:${RABBITMQ_PASSWORD}@${RABBITMQ_HOST}:5672/${RABBITMQ_VHOST}"
queue = "vector"
consumer = "vector"

[transforms.parse_json]
type = "remap"
inputs = ["rabbitmq"]
source = '''
  parsed, err = parse_json(.message)
  
  . = {
    "event": .routing,
    "timestamp": parsed.timestamp,
    "user_id": parsed.user_id,
    "user_login": parsed.user_login,
    "endpoint": parsed.endpoint,
    "parameters": parsed.parameters,
    "path_parameters": parsed.path_parameters,
    "metadata": parsed.metadata,
    "response_status_code": parsed.response_status_code,
  }

'''

[sinks.file]
type = "file"
inputs = ["parse_json"]
path = "/etc/vector/log/audit-%Y-%m-%d.log"
encoding.codec = "json"

[sinks.console]
type = "console"
inputs = ["parse_json"] 
encoding.codec = "json"

Значения переменным RABBITMQ_USER, RABBITMQ_PASSWORD, RABBITMQ_HOST и RABBITMQ_VHOST присваиваются в файле .env.

Настройка RabbitMQ для vector​

В RabbitMQ необходимо создать очередь vector. В веб интерфейсе RabbitMQ сделать это можно следующим образом: на вкладке Queues выберите Add a new queue и задайте следующие параметры:

  • Type: Classic;
  • Name: vector;
  • Durability: Durable;
  • Auto delete: No.

Выберите Add queue, будет создана очередь vector.

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

  • From exchange: pamir_audit_exchange;
  • Routing key: #.

Выберите Bind, маршрутизация настроена.

Routing key (Ключ маршрутизации) - это строка, которая поддерживает символы подстановки:

  • * - заменяет одно слово;
  • # - заменяет несколько слов или их отсутствие.

Перечень ключей маршрутизации приведён в документе Действия в системе.

Принцип работы vector​

RabbitMQ принимает сообщения Памир. vector подключается к RabbitMQ, читает сообщения из очереди vector, разбирает их согласно transforms.parse_json и сохраняет их в файл ./data/vector/log/audit-%Y-%m-%d.log в формате JSON.