> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cognisafe.uk/llms.txt
> Use this file to discover all available pages before exploring further.

# SIEM Forwarding

> Forward Cognisafe threat events to any SIEM via webhooks, syslog, or direct integration

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

<CardGroup cols={3}>
  <Card title="Outbound Webhooks" icon="webhook">
    Sub-second delivery of structured JSON. Recommended for real-time SOC alerting.
  </Card>

  <Card title="Log Shipping" icon="logs">
    Fluentd or Vector tail API pod stdout. Works with any SIEM that has a log collector agent.
  </Card>

  <Card title="PostgreSQL Direct" icon="database">
    Point your SIEM's JDBC/ODBC connector or a scheduled ETL at the `safety_scores` table.
  </Card>
</CardGroup>

***

## Method 1: Outbound Webhooks (Recommended)

### Configure in the Dashboard

<Steps>
  <Step title="Open Webhook Settings">
    Navigate to **Settings → Webhooks → Add Endpoint**.
  </Step>

  <Step title="Enter Target URL">
    Provide the HTTPS URL your SIEM or middleware exposes as a receiver. The URL must return HTTP 2xx within 10 seconds.
  </Step>

  <Step title="Select Event Types">
    Choose one or more: `threat_detected`, `usage_alert`, `redteam_complete`.
  </Step>

  <Step title="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.
  </Step>
</Steps>

### Event Types and Payload Schemas

#### `threat_detected`

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

```json theme={null}
{
  "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.

```json theme={null}
{
  "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.

```json theme={null}
{
  "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:**

```python theme={null}
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:**

```javascript theme={null}
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:**

```go theme={null}
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

```bash theme={null}
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:

```json theme={null}
{
  "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

```xml theme={null}
<!-- 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

```xml theme={null}
<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.

```toml theme={null}
# 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.

### Recommended Read Replica Setup

```bash theme={null}
# On RDS: create a read replica via AWS CLI
aws rds create-db-instance-read-replica \
  --db-instance-identifier cognisafe-read \
  --source-db-instance-identifier cognisafe-primary \
  --db-instance-class db.t4g.medium \
  --publicly-accessible false

# Grant a read-only SIEM user
psql $POSTGRES_URL -c "
  CREATE ROLE siem_reader WITH LOGIN PASSWORD 'strong-password-here';
  GRANT CONNECT ON DATABASE cognisafe TO siem_reader;
  GRANT USAGE ON SCHEMA public TO siem_reader;
  GRANT SELECT ON llm_requests, safety_scores, subscriptions TO siem_reader;
"
```

### Sample Security Dashboard Query

```sql theme={null}
-- 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;
```

```sql theme={null}
-- 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:

```sql theme={null}
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;
```