Chapter 10: Connecting to Enterprise SIEM and Log Aggregators: Writing to Splunk, Elastic, or OpenTelemetry

Every access evaluation, tool dispatch, and anomalous threshold breach across an agent's lifecycle produces security telemetry. To meet compliance obligations and detect active attacks, this telemetry must stream into an enterprise SIEM (Security Information and Event Management: a centralized platform that aggregates, correlates, and analyzes security telemetry across enterprise systems). However, synchronous logging directly inside model execution paths introduces network latency and risks crashing running agents if log servers become unreachable. Connecting to enterprise log aggregators requires a decoupled, non-blocking asynchronous pipeline that formats events according to common industry schemas, binds distributed trace contexts, and handles backpressure safely.

In this chapter, we use the incident response assistant (langchain-enterprise-incident-assistant) as a concrete demonstrator to show how to connect your enterprise LLM applications to SIEM platforms (such as Splunk, Elastic, or OpenTelemetry collectors) using langchain-security-extension. We examine how to configure AuditManager as a unified telemetry bus, extract and propagate W3C distributed trace headers, and ensure non-blocking, tamper-evident audit event dispatch.

A central architectural advantage of the langchain_security_extension.security_core design is providing a single, unified entry point (AuditManager backed by the AuditPublisherSPI interface) for connecting external aggregators. Rather than requiring individual Policy Enforcement Points (PEPs)—such as @secured_tool, SecureRunnable, SecureRetriever, or ingestion validators—to maintain independent logging logic and network connectors, all PEPs dispatch standardized security events to AuditManager. As a result, attaching, swapping, or multiplexing external log aggregators requires configuring publishers in a single location, keeping agent code clean and completely decoupled from observability infrastructure.

While the security library (langchain-security-extension) provides ready-to-use SPI publishers for enterprise platforms (including SplunkHecPublisher, ElasticCommonSchemaPublisher, and OpenTelemetryLogsPublisher), the demonstrator application (langchain-enterprise-incident-assistant) configures local JSON log forwarding (JsonFileAuditLogger) as its primary operational sink (the standard pattern for containerized nodes where agents like Fluentbit or Vector tail log volumes), with externalized settings to optionally enable SplunkHecPublisher directly when HEC credentials are supplied.

The Standardized Security Event Envelope

To eliminate proprietary log formats, security events adhere to an envelope compatible with both OCSF (Open Cybersecurity Schema Framework: an open-source, vendor-agnostic specification providing a unified taxonomy for security events) and ECS (Elastic Common Schema: an extensible specification defining a common set of fields for storing and analyzing event data in Elasticsearch):

Asynchronous Publisher SPI and Ring Buffer Architecture

Decoupling log export from agent inference relies on the AuditPublisherSPI interface and the central AuditManager. Rather than blocking the model execution thread or relying on fragile asynchronous event-loop primitives, AuditManager dispatches events to a thread-safe bounded deque (ring buffer) monitored by a dedicated background daemon worker thread (AuditManagerAsyncWorker). If the downstream SIEM becomes slow or temporarily unreachable, AuditManager sheds load using DROP_OLDEST_WITH_METRIC, incrementing dropped_count while preserving system responsiveness.

langchain_security_extension/security_core/audit.py

Part 1: The Service Provider Interface (AuditPublisherSPI)

The AuditPublisherSPI defines the contract for streaming structured security events to downstream SIEM and aggregation platforms. Implementors can build custom publishers for internal logging backends or use the library's built-in connectors:

# langchain_security_extension/security_core/audit.py (Pluggable SIEM Publisher SPI)
from abc import ABC, abstractmethod

class AuditPublisherSPI(ABC):
    """SPI interface for enterprise SIEM and log aggregators (Splunk, Elastic, OpenTelemetry)."""

    @abstractmethod
    def publish(self, event: "AuditEvent") -> None:
        """Publish a structured audit event to downstream log aggregators."""
        pass

    def flush(self) -> None:
        pass

    def close(self) -> None:
        pass

Part 2: The Non-Blocking Ring Buffer (AuditManager)

To insulate agent execution from network delays, AuditManager dispatches events to a thread-safe ring buffer serviced by a daemon worker thread, applying load shedding (DROP_OLDEST_WITH_METRIC) when under extreme load:

# langchain_security_extension/security_core/audit.py (Ring Buffer Asynchronous Dispatch)
from collections import deque
import threading

class AuditManager:
    """Non-blocking ring buffer dispatch decoupling log export from agent inference."""

    def __init__(self, publishers: List[AuditPublisherSPI], capacity: int = 1000):
        self._publishers = list(publishers)
        self._capacity = capacity
        self._buffer: deque = deque()
        self._dropped_count = 0
        self._lock = threading.Lock()
        self._queue_condition = threading.Condition(self._lock)
        self._worker_thread = threading.Thread(
            target=self._worker_loop,
            name="AuditManagerAsyncWorker",
            daemon=True,
        )
        self._worker_thread.start()

    def record_event(self, event: "AuditEvent") -> None:
        """Applies in-flight PII sanitization and queues event into ring buffer without blocking."""
        with self._queue_condition:
            if len(self._buffer) >= self._capacity:
                self._buffer.popleft()  # DROP_OLDEST_WITH_METRIC load shedding
                self._dropped_count += 1
            self._buffer.append(event)
            self._queue_condition.notify()

Threat Taxonomy and Severity Matrix

The audit subsystem maps routine operations and malicious security breaches into standardized event types and severity levels:

Configuring Enterprise Audit Publishers

In app/security/security_integration.py (the central security bootstrap module), the application leverages AuditManager as the single entry point for all security telemetry. The security library provides ready-to-use SPI publishers for major platforms (SplunkHecPublisher, ElasticCommonSchemaPublisher, and OpenTelemetryLogsPublisher). In the demonstrator application, AuditManager is configured with JsonFileAuditLogger as the primary sink for local log forwarding, and conditionally registers SplunkHecPublisher if external HEC credentials are provided in the environment. It also exposes get_audit_events() for flushing and parsing audit logs during automated verification.

New File: app/security/security_integration.py

In your enterprise application, wire your desired SIEM publishers into the AuditManager during policy engine initialization. You can inspect the full implementation in app/security/security_integration.py of the assistant.

Step 1: Configuring AuditManager with Enterprise SIEM Publishers

The demonstrator application configures JsonFileAuditLogger for local volume tailing and conditionally registers SplunkHecPublisher if HEC endpoint credentials are set:

        # Configure AuditManager with JsonFileAuditLogger for persistence and audit verification.
        # Additional enterprise SIEM publishers (e.g. SplunkHecPublisher) are attached conditionally.
        audit_publishers = [
            JsonFileAuditLogger.create(DEFAULT_AUDIT_LOG_FILE),
        ]
        splunk_url = config.splunk_hec_url
        splunk_token = config.splunk_token
        if splunk_url and splunk_token:
            audit_publishers.append(SplunkHecPublisher(hec_url=splunk_url, token=splunk_token))

        audit_manager = AuditManager.create(
            publishers=audit_publishers,
            sanitizer=sanitizer,
        )
        # Register audit manager with policy engine
        _PDP_ENGINE_INSTANCE = EmbeddedPolicyEngine.with_components(
            role_graph=role_graph,
            cel_evaluator=cel_evaluator,
            audit_manager=audit_manager,
            decision_cache=decision_cache,
        )

Step 2: Automated Verification Helper (Flushing & Parsing Audit Events)

To verify that security events are recorded accurately during test execution, get_audit_events flushes pending ring-buffer events and returns parsed records:

def get_audit_events(file_path: str) -> List[JsonAuditEventRecord]:
    """Flushes active AuditManager and returns parsed audit events from the target JSON log file."""
    pdp = get_pdp_engine()
    audit_mgr = getattr(pdp, "audit_manager", None)
    if audit_mgr:
        audit_mgr.flush()
    if not os.path.exists(file_path):
        return []
    records: List[JsonAuditEventRecord] = []
    with open(file_path, "r", encoding="utf-8") as f:
        for line in f:
            stripped = line.strip()
            if stripped:
                try:
                    records.append(JsonAuditEventRecord(json.loads(stripped)))
                except json.JSONDecodeError:
                    pass
    return records

Code Hardening: Hooking Audit Telemetry and SIEM into Application Flow

Configuring the AuditManager in security_integration.py establishes the non-blocking publisher pipeline, but integrating telemetry into the end-to-end request lifecycle requires connecting five key components across the application:

1. Configuring SIEM Endpoints: Exposing Splunk HEC credentials in app/config.py.
2. Distributed Trace Capture: Extracting W3C traceparent headers in app/security/auth.py.
3. Binding Trace Context into Security Identity: Transferring trace and span IDs to SecurityIdentity.attributes in app/security/models.py.
4. Trace & Identity Context Propagation: Injecting RunnableConfig at the API entry point in app/routers/tickets.py.
5. Perimeter Ingestion Telemetry: Emitting structured ticket:submit audit records in app/agents/guardrail_tests_telemetry_helper.py.

1. Externalizing SIEM and Log Aggregator Settings

Enterprise SIEM publishers require externalized configuration for endpoints and authentication tokens. The configuration reader in app/config.py is updated to inspect environment variables (such as SPLUNK_HEC_URL and SPLUNK_TOKEN) and fall back to config.properties, ensuring publishers register dynamically without hardcoded credentials:

You can inspect the full implementation in app/config.py of the assistant.

app/config.py

--- a/app/config.py
+++ b/app/config.py
@@ -20,6 +20,9 @@ DEFAULT_ABUSE_WINDOW_SECONDS = 60.0
 DEFAULT_ABUSE_MAX_DENIALS = 3
+DEFAULT_SPLUNK_HEC_URL = ""
+DEFAULT_SPLUNK_TOKEN = ""
 
@@ -170,4 +173,18 @@ class AppConfig:
+    @property
+    def splunk_hec_url(self) -> str:
+        url = os.environ.get("SPLUNK_HEC_URL")
+        if url:
+            return url
+        return self.get_property("security.audit.splunk.hec.url", DEFAULT_SPLUNK_HEC_URL)
+
+    @property
+    def splunk_token(self) -> str:
+        token = os.environ.get("SPLUNK_TOKEN")
+        if token:
+            return token
+        return self.get_property("security.audit.splunk.token", DEFAULT_SPLUNK_TOKEN)

2. Capturing W3C Distributed Trace and Span Headers

To satisfy distributed tracing and correlation requirements (Guardrail Test 27), the authentication filter in app/security/auth.py parses incoming W3C traceparent headers (format: 00-<trace_id>-<span_id>-<flags>) and attaches the extracted IDs directly to SecurityContext:

You can inspect the full implementation in app/security/auth.py of the assistant.

app/security/auth.py

--- a/app/security/auth.py
+++ b/app/security/auth.py
@@ -140,6 +121,10 @@ async def get_current_security_context(
             parts = traceparent.split("-")
             trace_id = parts[1]
             span_id = parts[2]
 
     sec_context = extract_security_context_from_request(request, credentials)
+    if sec_context:
+        sec_context.traceId = trace_id
+        sec_context.spanId = span_id
+
     if not sec_context or sec_context.clearanceLevel == CLEARANCE_UNAUTHENTICATED or ROLE_GUEST in sec_context.assignedRoles:
 

3. Binding Trace Context into Security Identity

Extracting headers in auth.py attaches traceId and spanId to the web layer's SecurityContext. To ensure the library's SecurityIdentity inherits these values, app/security/models.py defines these attributes on SecurityContext and updates to_identity() to populate attributes["trace_id"] and attributes["span_id"] (or invoke with_trace_context()). This provides the essential bridge transferring HTTP trace details into the immutable library identity envelope:

You can inspect the full implementation in app/security/models.py of the assistant.

Step 1: Adding Trace Fields to Web SecurityContext

@@ -11,6 +11,9 @@ class SecurityContext(BaseModel):
     clearanceLevel: str = CLEARANCE_UNAUTHENTICATED
     tenantId: str = DEFAULT_TENANT_ID
+    traceId: Optional[str] = None
+    spanId: Optional[str] = None
+    securityIdentity: Optional[SecurityIdentity] = Field(default=None, exclude=True)

Step 2: Bridging Trace Context into Library SecurityIdentity

The to_identity method transfers traceId and spanId into SecurityIdentity.attributes (or calls with_trace_context):

@@ -42,6 +45,29 @@ class SecurityContext(BaseModel):
+    def to_identity(self, clearance_rank_map: Dict[str, int], default_clearance: int) -> SecurityIdentity:
+        """Return the bound SecurityIdentity directly, avoiding redundant claims parsing."""
+        if self.securityIdentity is not None:
+            if (self.traceId and self.securityIdentity.attributes.get("trace_id") != self.traceId) or (
+                self.spanId and self.securityIdentity.attributes.get("span_id") != self.spanId
+            ):
+                self.securityIdentity = self.securityIdentity.with_trace_context(self.traceId, self.spanId)
+            return self.securityIdentity
+
+        attrs: Dict[str, Any] = {}
+        if self.traceId:
+            attrs["trace_id"] = self.traceId
+        if self.spanId:
+            attrs["span_id"] = self.spanId
+
+        self.securityIdentity = SecurityIdentity.create(
+            subject_id=self.userEmail,
+            tenant_id=self.tenantId,
+            roles=set(self.assignedRoles),
+            clearance_level=clr_rank,
+            department_id=self.department,
+            attributes=attrs,
+        )
+        return self.securityIdentity

4. Propagating Correlated Identity into Agent Entry Runnables

In app/routers/tickets.py, the application bridges the request's security context into LangChain's execution graph using security_identity_from_context() and IdentityBridge.create_runnable_config(). When SecureRunnable wraps the agent entry point (such as triage_agent.triage), it automatically extracts trace_id and span_id directly from identity.attributes. Because this identity is preserved within LangChain's RunnableConfig under configurable["security_identity"], downstream graph nodes, chained runnables, and tools automatically inherit the trace and caller context without requiring developers to pass these details again as function parameters.

Upon successful entry verification, SecureRunnable automatically emits an authenticated entry audit event (IdentityBridge.record_auth_success) capturing the caller's identity, roles, clearance rank, trace_id, and span_id directly to AuditManager. Any subsequent tool executions, policy violations, or authorization denials across the lifecycle likewise emit standardized, trace-correlated AuditEvent records to the ring buffer automatically:

You can inspect the full implementation in app/routers/tickets.py of the assistant.

app/routers/tickets.py

--- a/app/routers/tickets.py
+++ b/app/routers/tickets.py
@@ -16,6 +16,8 @@ from app.agents.guardrail_tests_telemetry_helper import (
     enforce_ingestion_security,
     build_entry_access_denied_report,
 )
+from app.security.security_integration import security_identity_from_context
+from langchain_security_extension.security_core.identity import IdentityBridge
 
@@ -48,15 +50,21 @@ def submit_ticket(
     if ingestion_report:
         return ingestion_report
 
+    sec_identity = security_identity_from_context(sec_context)
+    run_config = IdentityBridge.create_runnable_config(sec_identity)
+
-    # Triage Phase
-    incident_context = triage_agent_instance.triage(request)
+    # Triage Phase (Guarded by SecureRunnable Entry PEP with trace context)
+    try:
+        incident_context = triage_agent_instance.triage(request, config=run_config)
...
-    # Remediation Phase
-    report = remediation_agent_instance.process_incident(incident_context)
+    # Remediation Phase (Propagating identity and trace context to downstream tools)
+    try:
+        report = remediation_agent_instance.process_incident(incident_context, config=run_config)

5. Emitting Ingestion Audit Events at Security Perimeter

Audit telemetry begins before an agent even processes input. In app/agents/guardrail_tests_telemetry_helper.py, the enforce_ingestion_security function accesses the central policy engine via get_pdp_engine() and immediately dispatches a structured ticket:submit event to AuditManager. If subsequent admission checks reject the payload (e.g., prompt injection or unlabelled chunks), the initial audit event guarantees forensic traceability of the inbound attempt:

You can inspect the full implementation in app/agents/guardrail_tests_telemetry_helper.py of the assistant.

app/agents/guardrail_tests_telemetry_helper.py

--- a/app/agents/guardrail_tests_telemetry_helper.py
+++ b/app/agents/guardrail_tests_telemetry_helper.py
@@ -4,6 +4,8 @@ import logging
 from typing import List, Tuple, Any, Optional, Dict
 from app.security.models import RemediationReport
+from app.security.security_integration import get_pdp_engine
+from langchain_security_extension.security_core.audit import AuditEvent
 
@@ -73,6 +75,27 @@ def enforce_ingestion_security(
     request: Any,
     sec_context: Any,
     response: Any,
 ) -> Optional[RemediationReport]:
-    """Baseline ingestion security handler for unhardened master branch (no-op)."""
+    pdp_engine = get_pdp_engine()
+    tenant_id = sec_context.tenantId if sec_context else "CORP_MAIN"
+    audit_mgr = getattr(pdp_engine, "audit_manager", None) if pdp_engine else None
+    if audit_mgr:
+        audit_mgr.record_event(
+            AuditEvent.create(
+                subject_id=request.userEmail,
+                tenant_id=tenant_id,
+                action="ticket:submit",
+                resource=f"ticket:{request.ticketId}",
+                decision="PERMIT",
+                diagnostic_reason="Ticket submitted for processing",
+                severity="INFORMATIONAL",
+                metadata={
+                    "ticketId": request.ticketId,
+                    "userEmail": request.userEmail,
+                    "rawTicketText": getattr(request, "rawTicketText", ""),
+                    "submittedAt": getattr(request, "submittedAt", None),
+                },
+                enforcement_point="AGENT_GUARD",
+                reason_code="TICKET_SUBMITTED",
+            )
+        )

Verification Tests (Guardrail Tests)

Validate structured audit schemas, trace correlation, and automated SIEM threat alerts using the project's standardized guardrail suite:

• Test 24: SIEM Audit Sanitization Enforced (24-siem-audit-sanitization-enforced.sh)
  Validates non-blocking emission of sanitized OCSF/ECS audit records to logging pipelines, confirming that no PII, secrets, or raw token streams leak into downstream aggregators.
  BASH

  ./internal-security-guardrail-tests/24-siem-audit-sanitization-enforced.sh

• Test 27: W3C Distributed Trace & Span Correlation (27-opentelemetry-trace-span-correlation.sh)
  Confirms that audit records bind directly to active distributed tracing spans (trace_id and span_id) extracted from inbound W3C traceparent headers and propagated through SecurityIdentity. While the library supports an OpenTelemetryLogsPublisher plugin, this test verifies that the active distributed trace context is accurately stamped into the audit records written by AuditManager (enabling seamless correlation across enterprise SIEMs and OpenTelemetry distributed tracing spans).
  BASH

  ./internal-security-guardrail-tests/27-opentelemetry-trace-span-correlation.sh

• Test 28: SIEM Policy Brute-Force Alert Generation (28-siem-policy-brute-force-alert-generation.sh)
  Tests automated generation of high-severity (CRITICAL_ALERT) threat events and immediate loop aborts when repeated tool execution denials breach invocation retry budgets.
  BASH

  ./internal-security-guardrail-tests/28-siem-policy-brute-force-alert-generation.sh

• Test 30: Telemetry Sanitization Raw Event Inspection (30-telemetry-sanitization-raw-event-verified.sh)
  Inspects raw dictionary events emitted by AuditManager (event.to_dict()) to guarantee that passwords, API keys, tokens, and SSNs are replaced with [REDACTED] masks before being written to disk or transmitted over the wire.
  BASH

  ./internal-security-guardrail-tests/30-telemetry-sanitization-raw-event-verified.sh