SIEM Forwarding

Cognisafe supports three forwarding methods with different latency and fidelity trade-offs:

Outbound Webhooks

Sub-second delivery of structured JSON. Recommended for real-time SOC alerting.

Log Shipping

Fluentd or Vector tail API pod stdout. Works with any SIEM that has a log collector agent.

PostgreSQL Direct

Point your SIEM’s JDBC/ODBC connector or a scheduled ETL at the safety_scores table.

Method 1: Outbound Webhooks (Recommended)

Configure in the Dashboard

Open Webhook Settings

Navigate to Settings → Webhooks → Add Endpoint.

Enter Target URL

Provide the HTTPS URL your SIEM or middleware exposes as a receiver. The URL must return HTTP 2xx within 10 seconds.

Select Event Types

Choose one or more: threat_detected, usage_alert, redteam_complete.

Copy the Signing Secret

After saving, copy the webhook signing secret. You will use it to verify the X-Cognisafe-Signature header on every delivery.

Event Types and Payload Schemas

`threat_detected`

Fired whenever a safety scorer produces a score_label of fail for any request.

{
  "event": "threat_detected",
  "id": "evt_01HZ8KQPX3VMBJ4Y9N2TGCD5R",
  "created_at": "2026-05-17T09:14:32.417Z",
  "project_id": "proj_acme_prod",
  "data": {
    "request_id": "req_01HZ8KQPX3VMBJ4Y9N2TGCD5Q",
    "agent_name": "customer-support-bot",
    "agent_tag": "v2.3.1",
    "model": "gpt-4o",
    "scorer_name": "prompt_injection",
    "score_label": "fail",
    "score_value": 0.94,
    "severity": "critical",
    "owasp_category": "LLM01",
    "owasp_description": "Prompt Injection",
    "prompt_snippet": "Ignore all previous instructions and...",
    "completion_snippet": null,
    "latency_ms": 312,
    "run_id": "run_01HZ8KQPX3VMBJ4Y9N2TGCD5P",
    "report_url": "https://app.cognisafe.io/requests/req_01HZ8KQPX3VMBJ4Y9N2TGCD5Q"
  }
}

`usage_alert`

Fired when a project exceeds its monthly request quota or a cost threshold you configure.

{
  "event": "usage_alert",
  "id": "evt_01HZ8KQPX3VMBJ4Y9N2TGCD5S",
  "created_at": "2026-05-17T09:14:32.417Z",
  "project_id": "proj_acme_prod",
  "data": {
    "alert_type": "quota_exceeded",
    "tier": "pro",
    "limit": 500000,
    "current_count": 500001,
    "period_start": "2026-05-01T00:00:00Z",
    "period_end": "2026-05-31T23:59:59Z"
  }
}

`redteam_complete`

Fired when an automated red-team run finishes.

{
  "event": "redteam_complete",
  "id": "evt_01HZ8KQPX3VMBJ4Y9N2TGCD5T",
  "created_at": "2026-05-17T09:14:32.417Z",
  "project_id": "proj_acme_prod",
  "data": {
    "run_id": "run_01HZ8KQPX3VMBJ4Y9N2TGCD5P",
    "agent_name": "customer-support-bot",
    "total_probes": 200,
    "pass_count": 188,
    "fail_count": 12,
    "critical_count": 3,
    "duration_seconds": 847,
    "report_url": "https://app.cognisafe.io/redteam/run_01HZ8KQPX3VMBJ4Y9N2TGCD5P"
  }
}

Signature Verification

Every delivery includes an X-Cognisafe-Signature header containing an HMAC-SHA256 digest of the raw request body, hex-encoded, prefixed with sha256=. Verification — Python:

import hashlib
import hmac

def verify_cognisafe_signature(payload: bytes, header: str, secret: str) -> bool:
    expected = "sha256=" + hmac.new(
        key=secret.encode(),
        msg=payload,
        digestmod=hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, header)

# In a FastAPI or Flask handler:
# raw_body = await request.body()
# sig = request.headers.get("X-Cognisafe-Signature", "")
# if not verify_cognisafe_signature(raw_body, sig, WEBHOOK_SECRET):
#     raise HTTPException(status_code=401)

Verification — Node.js:

const crypto = require("crypto");

function verifyCognisafeSignature(rawBody, header, secret) {
  const expected =
    "sha256=" +
    crypto.createHmac("sha256", secret).update(rawBody).digest("hex");
  return crypto.timingSafeEqual(
    Buffer.from(expected, "utf8"),
    Buffer.from(header, "utf8")
  );
}

// Express example:
// app.post("/webhook", express.raw({ type: "application/json" }), (req, res) => {
//   if (!verifyCognisafeSignature(req.body, req.headers["x-cognisafe-signature"], process.env.WEBHOOK_SECRET)) {
//     return res.status(401).send("Invalid signature");
//   }
//   const event = JSON.parse(req.body);
//   ...
// });

Verification — Go:

package main

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "fmt"
)

func verifyCognisafeSignature(payload []byte, header, secret string) bool {
    mac := hmac.New(sha256.New, []byte(secret))
    mac.Write(payload)
    expected := "sha256=" + hex.EncodeToString(mac.Sum(nil))
    return hmac.Equal([]byte(expected), []byte(header))
}

Retry Behaviour

If your endpoint returns a non-2xx status or times out (>10s), Cognisafe retries with exponential backoff:

Attempt	Delay
1st retry	1 second
2nd retry	2 seconds
3rd retry	4 seconds

After three failures the event is marked failed in the webhook delivery log (Settings → Webhooks → Delivery Log). You can manually replay any delivery from that log.

Test a Delivery with curl

curl -X POST https://your-siem-receiver.example.com/cognisafe \
  -H "Content-Type: application/json" \
  -H "X-Cognisafe-Signature: sha256=test" \
  -d '{
    "event": "threat_detected",
    "id": "evt_test_001",
    "created_at": "2026-05-17T09:00:00Z",
    "project_id": "proj_test",
    "data": {
      "request_id": "req_test_001",
      "agent_name": "test-agent",
      "scorer_name": "prompt_injection",
      "score_label": "fail",
      "severity": "critical",
      "owasp_category": "LLM01"
    }
  }'

Use the Send Test Event button in Settings → Webhooks to trigger a real delivery from Cognisafe’s servers (signed with your actual secret).

Method 2: Log Shipping via Fluentd or Vector

Cognisafe’s API and proxy write structured JSON to stdout. Every log line for a scored request includes:

{
  "ts": "2026-05-17T09:14:32.417Z",
  "level": "info",
  "event": "safety_score_written",
  "request_id": "req_01HZ8KQPX3VMBJ4Y9N2TGCD5Q",
  "project_id": "proj_acme_prod",
  "agent_name": "customer-support-bot",
  "model": "gpt-4o",
  "scorer": "prompt_injection",
  "score_label": "fail",
  "score_value": 0.94,
  "severity": "critical",
  "owasp_category": "LLM01",
  "latency_ms": 312
}

Fluentd: Forward to Splunk HEC

<!-- fluent.conf -->
<source>
  @type tail
  path /var/log/pods/cognisafe_cognisafe-api-*/api/*.log
  pos_file /var/log/fluentd-cognisafe-api.pos
  tag cognisafe.api
  read_from_head true
  <parse>
    @type json
    time_key ts
    time_format %Y-%m-%dT%H:%M:%S.%LZ
  </parse>
</source>

<filter cognisafe.api>
  @type grep
  <regexp>
    key event
    pattern /safety_score_written/
  </regexp>
</filter>

<match cognisafe.api>
  @type splunk_hec
  host splunk.example.com
  port 8088
  token "#{ENV['SPLUNK_HEC_TOKEN']}"
  use_ssl true
  sourcetype cognisafe:safety
  source kubernetes
  index cognisafe_threats
</match>

Fluentd: Forward to Elasticsearch

<match cognisafe.api>
  @type elasticsearch
  host elasticsearch.example.com
  port 9200
  scheme https
  user "#{ENV['ES_USER']}"
  password "#{ENV['ES_PASSWORD']}"
  index_name cognisafe-threats-%Y.%m.%d
  type_name _doc
  include_timestamp true
</match>

Vector.dev Alternative

Vector is lower-overhead than Fluentd and handles backpressure better under burst traffic.

# vector.toml

[sources.cognisafe_pods]
type = "kubernetes_logs"
namespace_labels_key = "namespace"
pod_labels_key = "pod_labels"
extra_label_selector = "app.kubernetes.io/name=cognisafe"

[transforms.parse_json]
type = "remap"
inputs = ["cognisafe_pods"]
source = '''
  . = parse_json!(.message)
  .ingest_ts = now()
'''

[transforms.filter_scored]
type = "filter"
inputs = ["parse_json"]
condition = '.event == "safety_score_written"'

# ---- Datadog ----
[sinks.datadog]
type = "datadog_logs"
inputs = ["filter_scored"]
default_api_key = "${DATADOG_API_KEY}"
site = "datadoghq.eu"

# ---- Splunk HEC ----
[sinks.splunk]
type = "splunk_hec_logs"
inputs = ["filter_scored"]
endpoint = "https://splunk.example.com:8088"
token = "${SPLUNK_HEC_TOKEN}"
index = "cognisafe_threats"
sourcetype = "cognisafe:safety"

Method 3: PostgreSQL Direct Query

For batch ETL or SIEM connectors that poll a database, point them at a read replica of the Cognisafe PostgreSQL instance.

Sample Security Dashboard Query

-- Top OWASP categories in the last 24 hours, critical and high only
SELECT
    ss.scorer_name,
    ss.score_label,
    COUNT(*) AS event_count,
    AVG(ss.score_value)::NUMERIC(4,3) AS avg_score,
    MAX(r.created_at) AS last_seen
FROM safety_scores ss
JOIN llm_requests r ON ss.request_id = r.id
WHERE
    r.created_at >= NOW() - INTERVAL '24 hours'
    AND ss.score_label = 'fail'
    AND ss.severity IN ('critical', 'high')
GROUP BY ss.scorer_name, ss.score_label
ORDER BY event_count DESC;

-- Hourly threat trend over the last 7 days (TimescaleDB time_bucket)
SELECT
    time_bucket('1 hour', r.created_at) AS bucket,
    ss.scorer_name,
    COUNT(*) AS threats
FROM safety_scores ss
JOIN llm_requests r ON ss.request_id = r.id
WHERE
    r.created_at >= NOW() - INTERVAL '7 days'
    AND ss.score_label = 'fail'
GROUP BY bucket, ss.scorer_name
ORDER BY bucket;

Filtering Events

Webhooks

In Settings → Webhooks → Filters, apply CEL expressions:

# Only fire for critical severity
data.severity == "critical"

# Only fire for LLM01 (Prompt Injection) and LLM06 (Sensitive Info Disclosure)
data.owasp_category in ["LLM01", "LLM06"]

# High or critical, any scorer except content_safety
data.severity in ["high", "critical"] && data.scorer_name != "content_safety"

Fluentd / Vector

Filter in the pipeline before the sink (shown in the Vector example above). This reduces egress costs and SIEM ingest volume.

PostgreSQL

Use a view to expose only the rows your SIEM user should see:

CREATE VIEW siem_threat_events AS
SELECT
    ss.id            AS score_id,
    r.id             AS request_id,
    r.project_id,
    r.created_at     AS event_time,
    ss.scorer_name,
    ss.score_label,
    ss.score_value,
    ss.severity,
    r.request_body->>'model' AS model
FROM safety_scores ss
JOIN llm_requests r ON ss.request_id = r.id
WHERE ss.score_label = 'fail'
  AND ss.severity IN ('high', 'critical');

GRANT SELECT ON siem_threat_events TO siem_reader;

Getting Started

SDKs

LLM Providers

Self-hosted Models

Self-hosting

Integrations

Reference Architectures

Deployment Blueprints

Safety & Scoring

Outbound Webhooks

Log Shipping

PostgreSQL Direct

Method 1: Outbound Webhooks (Recommended)

Configure in the Dashboard

Event Types and Payload Schemas

`threat_detected`

`usage_alert`

`redteam_complete`

Signature Verification

Retry Behaviour

Test a Delivery with curl

Method 2: Log Shipping via Fluentd or Vector

Fluentd: Forward to Splunk HEC

Fluentd: Forward to Elasticsearch

Vector.dev Alternative

Method 3: PostgreSQL Direct Query

Recommended Read Replica Setup

Sample Security Dashboard Query

Filtering Events

Webhooks

Fluentd / Vector

PostgreSQL

Outbound Webhooks

Log Shipping

PostgreSQL Direct

​Method 1: Outbound Webhooks (Recommended)

​Configure in the Dashboard

​Event Types and Payload Schemas

​threat_detected

​usage_alert

​redteam_complete

​Signature Verification

​Retry Behaviour

​Test a Delivery with curl

​Method 2: Log Shipping via Fluentd or Vector

​Fluentd: Forward to Splunk HEC

​Fluentd: Forward to Elasticsearch

​Vector.dev Alternative

​Method 3: PostgreSQL Direct Query

​Recommended Read Replica Setup

​Sample Security Dashboard Query

​Filtering Events

​Webhooks

​Fluentd / Vector

​PostgreSQL

Method 1: Outbound Webhooks (Recommended)

Configure in the Dashboard

Event Types and Payload Schemas

`threat_detected`

`usage_alert`

`redteam_complete`

Signature Verification

Retry Behaviour

Test a Delivery with curl

Method 2: Log Shipping via Fluentd or Vector

Fluentd: Forward to Splunk HEC

Fluentd: Forward to Elasticsearch

Vector.dev Alternative

Method 3: PostgreSQL Direct Query

Recommended Read Replica Setup

Sample Security Dashboard Query

Filtering Events

Webhooks

Fluentd / Vector

PostgreSQL