SIEM forwarding¶

Forward every preflight decision to your SIEM without leaving your process. Runs on a background thread, fire-and-forget, never blocks the SDK caller.

pip install zotniq[siem]

Quickstart¶

Wire a forwarder into the on_decision constructor arg:

from zotniq import Zotniq
from zotniq.siem import SplunkForwarder

client = Zotniq(
    api_key="zot_sk_...",
    on_decision=SplunkForwarder(
        url="https://splunk.acme.com:8088/services/collector",
        token="hec-token-here",
    ),
)

# Every preflight.check() call fires an async POST to Splunk.
result = client.preflight.check("...", destination="AI_TOOL")

Built-in forwarders¶

Splunk HEC¶

from zotniq.siem import SplunkForwarder

SplunkForwarder(
    url="https://splunk.acme.com:8088/services/collector",
    token="hec-token",
    source="zotniq-sdk",           # default
    sourcetype="zotniq:decision",  # default
    index="dlp_events",            # optional
    verify_ssl=True,               # set False for self-signed HEC
)

Auth via the Splunk <token> Authorization header per HEC docs. Batches events efficiently, retries transient failures.

Datadog Logs API v2¶

from zotniq.siem import DatadogForwarder

DatadogForwarder(
    api_key="dd-api-key",
    site="datadoghq.com",          # or datadoghq.eu, us3.datadoghq.com, etc.
    ddsource="zotniq",             # default
    service="chat-backend",        # your service name
    tags=["env:prod", "team:security"],
)

Multi-region: site maps to the correct intake host automatically.

Generic webhook¶

Any SIEM with a JSON ingestion endpoint (Cribl, Sumo Logic, Elastic HTTP JSON input, custom SOC pipelines):

from zotniq.siem import WebhookForwarder

WebhookForwarder(
    url="https://soc.acme.com/hooks/dlp",
    headers={"Authorization": "Bearer secret-token"},
    timeout=10.0,
)

Local file (ndjson)¶

For pipelines that tail files (older Splunk deployments, Vector, Filebeat):

from zotniq.siem import FileForwarder

FileForwarder(path="/var/log/zotniq/decisions.ndjson")

Thread-safe. Rotation is deliberately out of scope — use standard logrotate against the target file.

Custom forwarders¶

Any Callable[[PreflightResult, dict], None] works as on_decision:

from zotniq import Zotniq
from zotniq.types import PreflightResult

def my_forwarder(result: PreflightResult, ctx: dict) -> None:
    my_soc_client.emit(
        decision=result.decision.value,
        findings=len(result.findings),
        text_hash=ctx["text_hash"],
        request_id=ctx["request_id"],
    )

client = Zotniq(api_key="zot_sk_...", on_decision=my_forwarder)

Failures inside your callback are caught and logged at WARNING. The SDK caller is never affected.

What's in the event¶

The context dict passed to your forwarder carries decision metadata only — never raw content:

Field	Type	Example
`timestamp`	ISO8601 UTC	`2026-08-25T20:00:00+00:00`
`destination`	str	`AI_TOOL`
`mode_used`	str	`cloud` or `local`
`text_hash`	SHA256 hex	`abc123...` (for correlation, not decryption)
`text_length`	int	bytes
`api_key_fingerprint`	str	`...abc123` (last 6 chars)
`request_id`	str	server's `x-request-id` or local UUID
`sdk_version`	str	`0.1.0`

Plus the built-in forwarders wrap this in an event envelope with:

Field	Example
`event`	`"zotniq.preflight.decision"`
`decision`	`"ALLOWED_WITH_MASKING"`
`summary`	`"Masked before send (EMAIL)."`
`findings_count`	`2`
`finding_types`	`["EMAIL", "SSN"]`

Privacy invariant¶

Raw text and masked_text NEVER cross the hook boundary. If you need the raw content in your SIEM (for correlation with other events in your SOC), capture it in your own code before calling preflight.check — in your process, never touched by the SDK.

Overflow policy¶

The internal queue caps at 10,000 events. If your SIEM is slow enough that events pile up faster than they drain, the SDK drops the oldest event and logs a WARNING. The drop counter is available via client.siem_stats():

stats = client.siem_stats()
# {"delivered": 12345, "dropped": 0, "queued": 3}

Right thing to do if you see drops in production: increase your SIEM's ingestion capacity, or add a queue between the SDK and the SIEM (Kinesis, Kafka, etc.) and use a custom forwarder to push into that queue instead.

Server-side audit stream vs SDK-side forwarding¶

Two orthogonal forwarding surfaces:

Server-side audit stream (see Audit Stream guides) — configured in your Zotniq dashboard, forwards decisions from the server's preflight_audit table. Cross-tenant, works for every SDK/API call regardless of client.
SDK-side SIEM forwarder (this page) — configured in your app code, forwards decisions from inside your process. Zero-latency, works even in mode="local", no dashboard config needed.

Most customers use one or the other, not both. Pick based on where your SIEM ingestion pipeline lives.