Интеграция событий безопасности CodeGraph с SIEM

CodeGraph передаёт события безопасности в систему, с которой уже работает SOC. Аутентификация, обращения к LLM и действия DLP попадают в общий поток расследования. Поле request_id связывает событие с запросом; тип, время, серьёзность и сведения о пользователе помогают настроить поиск и оповещения.

Диспетчер отправляет одно событие в несколько обработчиков. Syslog, CEF и LEEF поддерживают UDP, TCP и TLS. Для каждого обработчика CodeGraph хранит результат последних попыток и состояние healthy или degraded.

Выберите формат и транспорт

Согласуйте с SOC адрес, порт, формат и правила индексации. Для TLS используйте DNS-имя, которое соответствует сертификату приёмника. Отправитель проверяет имя сервера и цепочку сертификата по стандартному хранилищу доверия.

Настройте отправку

Параметры находятся в security.siem файла config.yaml. В рабочей конфигурации укажите адрес и порт приёмника SOC. В следующем примере локальный TCP-приёмник слушает 127.0.0.1:15140; формат — Syslog, категория — LOCAL0 (16).

Сохраните пример как config.yaml в отдельной папке проверки. Используйте окружение Python с установленным исходным пакетом CodeGraph.

security:
  enabled: true
  siem:
    enabled: true
    syslog:
      enabled: true
      protocol: tcp
      host: 127.0.0.1
      port: 15140
      facility: 16
      app_name: codegraph
    cef:
      enabled: false
    leef:
      enabled: false
    buffer:
      max_size: 10000
      flush_interval_seconds: 5
      retry_attempts: 3
      retry_backoff_seconds: 2.0

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

Проверьте отправку и получение

Этот пример отправляет одно синтетическое событие security.alert через настоящий TCP-транспорт. Приёмник печатает строку Syslog, а отправитель — результат и request_id.

1. Сохраните следующий код как receive_siem_test.py в папке проверки.

"""Receive one local TCP Syslog message for the SIEM transport check."""
import socket

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.bind(("127.0.0.1", 15140))
    server.listen(1)
    server.settimeout(30)
    print("Listening on 127.0.0.1:15140", flush=True)
    connection, _ = server.accept()
    with connection, connection.makefile("r", encoding="utf-8") as stream:
        print(stream.readline().rstrip(), flush=True)

2. Сохраните отправитель как send_siem_test.py в той же папке.

"""Send a synthetic security event using the selected SIEM configuration."""
import json
from uuid import uuid4

from src.security.config import SecurityConfigSource, load_security_config
from src.security.siem.base_handler import SecurityEvent, SecurityEventType
from src.security.siem.dispatcher import SIEMDispatcher

config = load_security_config(SecurityConfigSource(path="config.yaml"))
event = SecurityEvent.create(
    event_type=SecurityEventType.SECURITY_ALERT,
    message="SIEM integration test event",
    request_id=str(uuid4()),
    severity=6,
    details={"source": "documentation-check", "test": True},
)
with SIEMDispatcher(config.siem) as dispatcher:
    results = dispatcher.dispatch_sync_detailed(event)
    print(json.dumps({"request_id": event.request_id, "handlers": results}))
    succeeded = bool(results) and all(item["dispatched"] for item in results)
raise SystemExit(0 if succeeded else 1)

3. В первом терминале запустите приёмник:

python receive_siem_test.py

Дождитесь строки Listening on 127.0.0.1:15140.

4. Во втором терминале, из той же папки, отправьте событие:

python send_siem_test.py

Отправитель завершается с кодом 0. В массиве handlers найдите dispatched: true. В строке приёмника должны быть event_type="security.alert", сообщение SIEM integration test event и тот же request_id, что в JSON отправителя. После получения одной строки приёмник завершает работу.

5. Повторите команду отправителя после завершения приёмника. Она завершается с кодом 1; обработчик возвращает dispatched: false и error_summary. Это проверяет диагностику недоступного TCP-получателя.

Для подключения SOC замените адрес и порт в рабочей конфигурации. Найдите событие по request_id в индексе SIEM и проверьте время, тип и уровень серьёзности. При UDP запись у получателя подтверждает приём; при TCP/TLS она также показывает результат индексации после сетевой отправки.

Как работает очередь

dispatch(event) ставит событие в очередь процесса. Фоновый поток отправляет накопленные события с периодом flush_interval_seconds. При сбое событие возвращается в очередь до следующей отправки. Общее число попыток задаёт retry_attempts.

Очередь вмещает max_size событий. При заполнении новое событие вытесняет самое старое, счётчик dropped увеличивается. При перезапуске процесса очередь создаётся заново. Учитывайте эти свойства при выборе размера очереди и режима обновления.

dispatch_sync(event) отправляет событие сразу и возвращает успешный результат хотя бы одного обработчика. dispatch_sync_detailed(event) возвращает отдельный результат каждого обработчика. flush() запускает отправку накопленной очереди.

Контролируйте интеграцию через API

Войдите в CodeGraph и используйте маршруты REST API:

Схемы запросов и ответов доступны в /api/docs развёрнутого API. Для изменения конфигурации выполните вход с ролью администратора.

Настройте эксплуатацию

Оповещайте SOC о состоянии degraded, росте очереди, повторных отправках и dropped. Задайте пороги по измеренной частоте событий и допустимой задержке приёма. Согласуйте правила обработки чувствительных полей, срок хранения и доступ к индексу.

При отказе сохраните error_summary, время и конфигурацию без секретов. Проверьте DNS, сеть, TLS и состояние приёмника. После восстановления отправьте очередь и сопоставьте её счётчики с событиями в SIEM. Для периода деградации зафиксируйте ответственного, дополнительный канал наблюдения и условие завершения восстановления.

Для приёмки сохраните конфигурацию, результаты синтетического события и проверки отказа, статистику очереди, запись в индексе SIEM и настройки оповещений. Проверку событий аутентификации и DLP проводите через соответствующие пользовательские действия: OAuth/LDAP и DLP.

Поля события

Событие содержит event_type, timestamp, request_id, message и severity. Дополнительные поля описывают пользователя, сеть, провайдера, модель и действие DLP. Значения severity соответствуют шкале Syslog 0–7.