Integrate CodeGraph security events with SIEM

Choose a format and transport

Agree on the address, port, format and indexing rules with the SOC. For TLS, use a DNS name that matches the receiver certificate. The sender validates the server name and certificate chain against the default trust store.

Configure delivery

Settings live under security.siem in config.yaml. Set the SOC receiver address and port in your deployment. The following example uses a local TCP receiver at 127.0.0.1:15140, Syslog and the LOCAL0 facility (16).

Save this example as config.yaml in a separate check directory. Use a Python environment with the CodeGraph source package installed.

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

Select handlers with enabled. To send both formats, configure both receivers. The dispatcher records each handler result.

Verify sending and receipt

This example sends one synthetic security.alert event over the real TCP transport. The receiver prints the Syslog line. The sender prints the result and request_id.

1. Save this code as receive_siem_test.py in the check directory.

"""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. Save the sender as send_siem_test.py in the same directory.

"""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. Start the receiver in the first terminal:

python receive_siem_test.py

Wait for Listening on 127.0.0.1:15140.

4. Send the event from the same directory in the second terminal:

python send_siem_test.py

The sender exits with code 0. Find dispatched: true in the handlers array. The receiver line contains event_type="security.alert", the message SIEM integration test event and the same request_id as the sender JSON. The receiver exits after printing one line.

5. Run the sender again after the receiver exits. It exits with code 1; the handler returns dispatched: false and an error_summary. This checks diagnostics for an unavailable TCP receiver.

For the SOC connection, set its address and port in the deployment configuration. Find the event by request_id in the SIEM index. Check its time, type and severity. For UDP, the receiver record confirms receipt. For TCP/TLS, it also shows indexing after the network send.

How buffering works

dispatch(event) enqueues the event in process memory. A background thread sends pending events at flush_interval_seconds intervals. Failed sends return to the queue for the next flush, up to the total number of attempts set by retry_attempts.

The queue holds max_size events. When full, a new event replaces the oldest one and increments dropped. A process restart creates a new queue. Use these properties to choose queue capacity and plan updates.

dispatch_sync(event) sends immediately and returns success when at least one handler sends the event. dispatch_sync_detailed(event) returns each handler result. flush() starts sending queued events.