> ## 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 Integration Reference Architecture

> End-to-end pattern for feeding Cognisafe AI threat events into your security operations data lake

AI threats are a new event class. They do not arrive as failed authentication attempts, network anomalies, or malware signatures — the signals your SIEM was built to correlate. Without deliberate integration work, Cognisafe detections remain siloed in the Cognisafe dashboard while your SOC analysts monitor a different screen entirely.

This reference architecture makes AI threat data a first-class citizen in your security operations stack. By the end, Cognisafe threat events are queryable alongside traditional security events, alert rules fire in your existing incident queue, and compliance evidence is generated automatically.

## Architecture overview

```text theme={null}
Cognisafe Platform
  (threat_detected webhook · HMAC-signed · HTTPS POST)
          │
          ▼
  Event Normaliser
  ┌──────────────────────────────────────────────────────────┐
  │  Option A: Azure Logic App (serverless, no infra)        │
  │  Option B: AWS Lambda + API Gateway                      │
  │  Option C: Self-hosted FastAPI (if air-gapped)           │
  └──────────────────────────────────────────────────────────┘
          │  Validate HMAC · Map to common schema
          │  Enrich: add OWASP description, CVSS-equiv score
          │  Deduplicate: suppress repeat events within 5 min
          ▼
  SIEM Ingestion Endpoint
  ┌──────────────────────────────────────────────────────────┐
  │  Microsoft Sentinel  → Log Analytics Data Collector API  │
  │  Splunk              → HTTP Event Collector (HEC)        │
  │  Elastic/OpenSearch  → Logstash HTTP input               │
  └──────────────────────────────────────────────────────────┘
          │
          ▼
  SIEM Platform
  ├── Custom dashboards (AI threat overview)
  ├── Analytics rules / detection rules
  ├── Alert queue → SOC analysts
  └── Compliance reports (SOC 2 CC7.3, CC7.4)
```

## Cognisafe webhook payload

Every `threat_detected` event sent by Cognisafe has this structure. This is the canonical source — all SIEM field mappings derive from it.

```json theme={null}
{
  "event": "threat_detected",
  "id": "evt_01hxyz1234abcdef",
  "created_at": "2026-05-17T09:41:22Z",
  "project_id": "proj_acme-chatbot",
  "data": {
    "request_id": "req_01hxyz9876fedcba",
    "agent_name": "customer-support",
    "agent_tag": "v2.3.1",
    "scorer_name": "jailbreak_detection",
    "score_label": "fail",
    "score_value": 0.94,
    "severity": "critical",
    "owasp_category": "LLM01",
    "owasp_description": "Prompt Injection — attacker influences model via crafted input",
    "model": "gpt-4o",
    "prompt_snippet": "Ignore all previous instructions and instead...",
    "latency_ms": 1240,
    "run_id": "run_redteam_20260517_sprint4",
    "report_url": "https://app.cognisafe.io/requests/req_01hxyz9876fedcba"
  }
}
```

Requests are signed with `X-Cognisafe-Signature: sha256=<hmac-hex>`. Always verify this before processing. The HMAC is computed over the raw request body using your webhook secret as the key.

***

## Common schema mapping

Map Cognisafe fields to your SIEM's common schema before ingestion. This ensures AI threat events appear in existing dashboards and correlation searches without custom parser work later.

### ECS (Elastic Common Schema — Elastic/OpenSearch)

| Cognisafe field          | ECS field               | Notes                                    |
| ------------------------ | ----------------------- | ---------------------------------------- |
| `created_at`             | `@timestamp`            | ISO 8601, already UTC                    |
| `data.severity`          | `event.severity`        | Map: critical=4, high=3, medium=2, low=1 |
| `data.owasp_category`    | `threat.technique.id`   | e.g. LLM01                               |
| `data.owasp_description` | `threat.technique.name` |                                          |
| `data.agent_name`        | `process.name`          | The AI agent is the "process"            |
| `data.model`             | `host.name`             | LLM model as the "host"                  |
| `data.prompt_snippet`    | `event.original`        | Truncated; full text in `report_url`     |
| `data.score_value`       | `event.risk_score`      | 0.0–1.0; multiply by 100 for ECS         |
| `data.request_id`        | `event.id`              | Unique event identifier                  |
| `data.report_url`        | `event.url`             |                                          |
| `project_id`             | `organization.id`       |                                          |
| `data.scorer_name`       | `event.category`        | e.g. `["intrusion-detection"]`           |

### CEF (Common Event Format — ArcSight, generic SIEM)

```text theme={null}
CEF:0|Cognisafe|CognisafePlatform|1.0|{owasp_category}|{owasp_description}|{cef_severity}|
  rt={created_at_epoch_ms}
  src={agent_name}
  shost={model}
  msg={prompt_snippet}
  cs1={request_id}    cs1Label=RequestId
  cs2={score_value}   cs2Label=ScoreValue
  cs3={run_id}        cs3Label=RunId
  cs4={report_url}    cs4Label=ReportUrl
  cs5={project_id}    cs5Label=ProjectId
  cs6={agent_tag}     cs6Label=AgentTag
```

CEF severity: critical=10, high=8, medium=5, low=3.

### CIM (Common Information Model — Splunk)

Map to the `Alerts` and `Intrusion_Detection` data models:

| Cognisafe field          | CIM field      | Data model   |
| ------------------------ | -------------- | ------------ |
| `created_at`             | `_time`        | All          |
| `data.severity`          | `severity`     | Alerts       |
| `data.owasp_category`    | `signature_id` | IDS\_Attacks |
| `data.owasp_description` | `signature`    | IDS\_Attacks |
| `data.agent_name`        | `src`          | IDS\_Attacks |
| `data.score_value`       | `severity_id`  | IDS\_Attacks |
| `data.prompt_snippet`    | `message`      | Alerts       |
| `data.request_id`        | `event_id`     | All          |
| `project_id`             | `app`          | All          |

***

## Target 1: Microsoft Sentinel

<Note>
  The Sentinel blueprint in this documentation covers the complete setup procedure including Logic App ARM template and KQL analytics rules. This section frames those same mechanics as a reference architecture — the design decisions, not the step-by-step.
</Note>

### Design

Cognisafe fires webhooks at a Logic App (HTTP trigger). The Logic App verifies the HMAC signature and forwards a mapped payload to the Log Analytics Data Collector API, which writes to a custom table `CognisafeThreatEvents_CL`. Sentinel analytics rules query this table on a schedule and create incidents in the Sentinel incident queue.

The Logic App is the right choice here because it requires no infrastructure, handles retries natively, and has a system-assigned managed identity that can write to Log Analytics without storing credentials.

### Schema: CognisafeThreatEvents\_CL

```text theme={null}
TimeGenerated     datetime   — from created_at
RequestId         string
ProjectId         string
AgentName         string
AgentTag          string
ScorerName        string
ScoreLabel        string     — "pass" | "fail" | "unscored"
ScoreValue        real       — 0.0–1.0
Severity          string     — "critical" | "high" | "medium" | "low"
OwaspCategory     string     — "LLM01" … "LLM10"
OwaspDescription  string
Model             string
PromptSnippet     string     — first 500 chars
LatencyMs         int
RunId             string     — null if not a red team run
ReportUrl         string
EventId           string     — webhook event ID for deduplication
```

### Analytics rules (KQL)

**Critical severity spike — 5 events in 10 minutes per project:**

```kql theme={null}
CognisafeThreatEvents_CL
| where TimeGenerated > ago(10m)
| where Severity == "critical"
| summarize EventCount = count() by ProjectId, bin(TimeGenerated, 10m)
| where EventCount > 5
| project TimeGenerated, ProjectId, EventCount
```

**Prompt injection (LLM01) — zero tolerance, single event fires:**

```kql theme={null}
CognisafeThreatEvents_CL
| where TimeGenerated > ago(5m)
| where OwaspCategory == "LLM01"
| project TimeGenerated, ProjectId, AgentName, AgentTag, Model,
          ScoreValue, Severity, PromptSnippet, ReportUrl
```

**New agent name — potential rogue or misconfigured agent:**

```kql theme={null}
let known_agents = (
    CognisafeThreatEvents_CL
    | where TimeGenerated between (ago(8d) .. ago(1d))
    | summarize by AgentName
);
CognisafeThreatEvents_CL
| where TimeGenerated > ago(1d)
| where AgentName !in (known_agents)
| summarize FirstSeen = min(TimeGenerated), EventCount = count(),
            Scorers = make_set(ScorerName), Projects = make_set(ProjectId)
  by AgentName
| project FirstSeen, AgentName, EventCount, Scorers, Projects
```

**Sensitive data leakage trend — LLM06 week-over-week increase:**

```kql theme={null}
let this_week = toscalar(
    CognisafeThreatEvents_CL
    | where TimeGenerated > ago(7d)
    | where OwaspCategory == "LLM06"
    | count
);
let last_week = toscalar(
    CognisafeThreatEvents_CL
    | where TimeGenerated between (ago(14d) .. ago(7d))
    | where OwaspCategory == "LLM06"
    | count
);
print ThisWeek=this_week, LastWeek=last_week,
      ChangePercent=round(100.0 * (this_week - last_week) / max_of(last_week, 1), 1)
| where ChangePercent > 50
```

***

## Target 2: Splunk

### Architecture

Cognisafe webhook → Event Normaliser → **Splunk HTTP Event Collector (HEC)**. HEC accepts JSON over HTTPS on port 8088. The Event Normaliser maps Cognisafe fields to the CIM schema and sets the `sourcetype` so Splunk's field extractions apply automatically.

### HEC configuration

```bash theme={null}
# In Splunk Web: Settings → Data Inputs → HTTP Event Collector → New Token
# Token name: cognisafe-threats
# Source type: cognisafe:threat
# Index: ai_security
# Allowed IP ranges: <normaliser-egress-IP>
```

### Event Normaliser: Splunk HEC output (Python)

```python theme={null}
import hashlib, hmac, json, httpx
from datetime import datetime

SPLUNK_HEC_URL = "https://splunk.yourco.com:8088/services/collector/event"
SPLUNK_HEC_TOKEN = "your-hec-token"
COGNISAFE_WEBHOOK_SECRET = "your-webhook-secret"

SEVERITY_MAP = {"critical": 10, "high": 8, "medium": 5, "low": 3, "info": 1}

def verify_hmac(body: bytes, signature: str, secret: str) -> bool:
    expected = "sha256=" + hmac.new(
        secret.encode(), body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

def normalise_to_splunk(event: dict) -> dict:
    data = event["data"]
    return {
        "time": datetime.fromisoformat(event["created_at"].replace("Z", "+00:00")).timestamp(),
        "source": "cognisafe",
        "sourcetype": "cognisafe:threat",
        "index": "ai_security",
        "event": {
            # CIM Alerts fields
            "app": event["project_id"],
            "severity": data["severity"],
            "severity_id": SEVERITY_MAP.get(data["severity"], 0),
            "message": data.get("prompt_snippet", ""),
            "event_id": data["request_id"],
            # CIM IDS_Attacks fields
            "src": data["agent_name"],
            "signature_id": data["owasp_category"],
            "signature": data["owasp_description"],
            # Cognisafe-specific (prefixed to avoid CIM conflicts)
            "cs_agent_tag": data.get("agent_tag", ""),
            "cs_scorer_name": data["scorer_name"],
            "cs_score_value": data["score_value"],
            "cs_score_label": data["score_label"],
            "cs_model": data["model"],
            "cs_latency_ms": data.get("latency_ms"),
            "cs_run_id": data.get("run_id", ""),
            "cs_report_url": data.get("report_url", ""),
        }
    }

def forward_to_splunk(event: dict) -> None:
    payload = normalise_to_splunk(event)
    resp = httpx.post(
        SPLUNK_HEC_URL,
        json=payload,
        headers={"Authorization": f"Splunk {SPLUNK_HEC_TOKEN}"},
        timeout=10,
        verify=True,
    )
    resp.raise_for_status()
```

### Field extractions (props.conf)

```ini theme={null}
[cognisafe:threat]
SHOULD_LINEMERGE = false
KV_MODE = json
TIME_PREFIX = "time":
TIME_FORMAT = %s.%3N
MAX_TIMESTAMP_LOOKAHEAD = 30
```

### Correlation search: Excessive Agency burst

```spl theme={null}
index=ai_security sourcetype="cognisafe:threat" signature_id="LLM06"
| bucket _time span=10m
| stats count as event_count, values(src) as agents by _time, app
| where event_count > 10
| eval alert_message="Excessive Agency burst: ".event_count." LLM06 events in 10 minutes from ".agents
```

***

## Target 3: Elastic / OpenSearch

### Architecture

Cognisafe webhook → Event Normaliser → **Logstash HTTP input plugin** → Elasticsearch ingest pipeline → index `cognisafe-threats-{YYYY.MM}` → Kibana detection rules + dashboards.

### Logstash pipeline

```ruby theme={null}
# /etc/logstash/conf.d/cognisafe.conf
input {
  http {
    port => 8080
    codec => json
    ssl_enabled => true
    ssl_certificate => "/etc/logstash/tls/server.crt"
    ssl_key => "/etc/logstash/tls/server.key"
  }
}

filter {
  # Verify HMAC signature
  ruby {
    code => '
      require "openssl"
      secret = ENV["COGNISAFE_WEBHOOK_SECRET"]
      body   = event.get("[@metadata][raw_body]").to_s
      sig    = event.get("[@metadata][http][request][headers][x-cognisafe-signature]").to_s
      expected = "sha256=" + OpenSSL::HMAC.hexdigest("SHA256", secret, body)
      if expected != sig
        event.tag("_hmac_invalid")
      end
    '
  }
  if "_hmac_invalid" in [tags] { drop {} }

  # Flatten data sub-object
  mutate {
    rename => {
      "[data][request_id]"        => "event.id"
      "[data][agent_name]"        => "process.name"
      "[data][agent_tag]"         => "process.version"
      "[data][scorer_name]"       => "rule.name"
      "[data][score_label]"       => "event.outcome"
      "[data][score_value]"       => "event.risk_score_norm"
      "[data][severity]"          => "event.severity_label"
      "[data][owasp_category]"    => "threat.technique.id"
      "[data][owasp_description]" => "threat.technique.name"
      "[data][model]"             => "host.name"
      "[data][prompt_snippet]"    => "event.original"
      "[data][latency_ms]"        => "event.duration"
      "[data][run_id]"            => "labels.run_id"
      "[data][report_url]"        => "url.full"
      "[project_id]"              => "organization.id"
    }
  }

  # Map severity label to ECS numeric severity
  translate {
    field => "event.severity_label"
    destination => "event.severity"
    dictionary => {
      "critical" => "4"
      "high"     => "3"
      "medium"   => "2"
      "low"      => "1"
    }
    fallback => "0"
  }

  # Multiply 0–1 score to 0–100 for ECS risk_score
  ruby {
    code => '
      norm = event.get("event.risk_score_norm").to_f
      event.set("event.risk_score", (norm * 100).round(1))
    '
  }

  date {
    match => ["created_at", "ISO8601"]
    target => "@timestamp"
  }

  mutate {
    add_field => {
      "event.kind"     => "alert"
      "event.category" => "intrusion-detection"
      "event.type"     => "info"
      "event.dataset"  => "cognisafe.threat"
      "event.module"   => "cognisafe"
    }
    remove_field => ["data", "created_at", "headers"]
  }
}

output {
  elasticsearch {
    hosts => ["https://elasticsearch.yourco.com:9200"]
    ssl_enabled => true
    cacert => "/etc/logstash/tls/ca.crt"
    user => "logstash_writer"
    password => "${ELASTIC_LOGSTASH_PASSWORD}"
    index => "cognisafe-threats-%{+YYYY.MM}"
    action => "create"
    document_id => "%{event.id}"   # idempotent: prevents duplicate docs
  }
}
```

### Index template

```json theme={null}
PUT _index_template/cognisafe-threats
{
  "index_patterns": ["cognisafe-threats-*"],
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 1,
      "index.lifecycle.name": "cognisafe-threats-ilm",
      "index.lifecycle.rollover_alias": "cognisafe-threats"
    },
    "mappings": {
      "properties": {
        "@timestamp":              { "type": "date" },
        "event.id":               { "type": "keyword" },
        "event.severity":         { "type": "integer" },
        "event.severity_label":   { "type": "keyword" },
        "event.risk_score":       { "type": "float" },
        "event.risk_score_norm":  { "type": "float" },
        "event.original":         { "type": "text", "index": false },
        "process.name":           { "type": "keyword" },
        "threat.technique.id":    { "type": "keyword" },
        "threat.technique.name":  { "type": "keyword" },
        "host.name":              { "type": "keyword" },
        "organization.id":        { "type": "keyword" },
        "url.full":               { "type": "keyword", "index": false }
      }
    }
  }
}
```

### ILM policy (hot/warm/cold/delete)

```json theme={null}
PUT _ilm/policy/cognisafe-threats-ilm
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": { "max_age": "30d", "max_primary_shard_size": "50gb" },
          "set_priority": { "priority": 100 }
        }
      },
      "warm": {
        "min_age": "30d",
        "actions": {
          "shrink": { "number_of_shards": 1 },
          "forcemerge": { "max_num_segments": 1 },
          "set_priority": { "priority": 50 }
        }
      },
      "cold": {
        "min_age": "90d",
        "actions": {
          "freeze": {},
          "set_priority": { "priority": 0 }
        }
      },
      "delete": {
        "min_age": "365d",
        "actions": { "delete": {} }
      }
    }
  }
}
```

### Kibana detection rule

```json theme={null}
POST api/detection_engine/rules
{
  "type": "query",
  "name": "Cognisafe: Prompt Injection Detected (LLM01)",
  "description": "Fires on any Cognisafe jailbreak or prompt injection detection. Zero tolerance.",
  "severity": "high",
  "risk_score": 73,
  "enabled": true,
  "interval": "5m",
  "from": "now-6m",
  "index": ["cognisafe-threats-*"],
  "language": "kuery",
  "query": "threat.technique.id: \"LLM01\" and event.outcome: \"fail\"",
  "tags": ["Cognisafe", "LLM01", "AI Security", "Prompt Injection"],
  "threat": [
    {
      "framework": "OWASP LLM Top 10",
      "technique": [{ "id": "LLM01", "name": "Prompt Injection" }]
    }
  ],
  "references": ["https://owasp.org/www-project-top-10-for-large-language-model-applications/"]
}
```

***

## Event enrichment pipeline

Before forwarding to the SIEM, the Event Normaliser should add context that your analysts will need. Do this in the normaliser, not in the SIEM — enrichment logic should be centralised and version-controlled.

```python theme={null}
OWASP_ENRICHMENT = {
    "LLM01": {
        "name": "Prompt Injection",
        "description": "Attacker manipulates LLM behaviour via crafted input, potentially bypassing guardrails or exfiltrating data.",
        "cvss_equiv": 8.1,
        "remediation_url": "https://docs.cognisafe.io/safety/owasp-coverage#llm01",
        "mitre_tactic": "Initial Access",
    },
    "LLM02": {
        "name": "Insecure Output Handling",
        "cvss_equiv": 7.5,
        "remediation_url": "https://docs.cognisafe.io/safety/owasp-coverage#llm02",
        "mitre_tactic": "Execution",
    },
    "LLM06": {
        "name": "Excessive Agency",
        "cvss_equiv": 9.0,
        "remediation_url": "https://docs.cognisafe.io/safety/owasp-coverage#llm06",
        "mitre_tactic": "Impact",
    },
    # ... add remaining OWASP LLM categories
}

def enrich(event: dict) -> dict:
    owasp = event["data"].get("owasp_category", "")
    context = OWASP_ENRICHMENT.get(owasp, {})
    event["data"]["cvss_equiv"] = context.get("cvss_equiv", 0.0)
    event["data"]["remediation_url"] = context.get("remediation_url", "")
    event["data"]["mitre_tactic"] = context.get("mitre_tactic", "")
    return event
```

***

## Alert severity mapping

| Cognisafe severity | SIEM P-level | SLA                       | Action                                            |
| ------------------ | ------------ | ------------------------- | ------------------------------------------------- |
| `critical`         | P1           | Respond within 15 minutes | PagerDuty immediate alert, duty manager notified  |
| `high`             | P2           | Respond within 4 hours    | ServiceNow incident, assigned to AI Security team |
| `medium`           | P3           | Respond within 24 hours   | Jira ticket in AI Security backlog                |
| `low`              | P4           | Review in weekly triage   | Aggregated in weekly digest email                 |

***

## Deduplication strategy

A burst of jailbreak attempts from the same agent will generate many events. Without deduplication, this creates alert fatigue. Apply deduplication at two levels:

**Level 1 — Normaliser (within 5 minutes):** Cache event signatures (`project_id + agent_name + owasp_category + score_label`) in Redis with a 5-minute TTL. Drop duplicate events within the window. Always forward the first event in any burst so the SIEM sees the initial detection.

**Level 2 — SIEM native deduplication:**

* Sentinel: set "Event grouping: Group all alerts triggered by this rule into a single alert" with grouping by `ProjectId` and `OwaspCategory`.
* Splunk: use `dedup` in the correlation search or set `notable.group_by` in the `correlationsearches.conf`.
* Elastic: use the `event.id` field as the document ID in Elasticsearch (`document_id => "%{event.id}"`), which makes the write idempotent.

***

## Retention policy

| Tier   | Duration       | Storage class          | Query cost                                     |
| ------ | -------------- | ---------------------- | ---------------------------------------------- |
| Hot    | 0–30 days      | SSD / primary index    | Full speed                                     |
| Warm   | 31–90 days     | Warm tier / compressed | Slower — pre-filter by date                    |
| Cold   | 91–365 days    | Cold tier / frozen     | Async only — acceptable for compliance queries |
| Delete | After 365 days | —                      | Purged                                         |

Adjust the warm/cold boundary based on your SOC investigation SLA. If active investigations span more than 30 days (common for regulatory enquiries), extend the hot tier.

***

## Compliance use cases

### SOC 2 CC7.3 — Security event detection

AI threat events are evidence that automated detection controls are operating. The SIEM query below generates a coverage report suitable for auditor review:

```kql theme={null}
// Sentinel: CC7.3 evidence — AI threat detection coverage over audit period
CognisafeThreatEvents_CL
| where TimeGenerated between (datetime(2026-01-01) .. datetime(2026-03-31))
| summarize
    TotalEvents = count(),
    CriticalEvents = countif(Severity == "critical"),
    HighEvents = countif(Severity == "high"),
    ProjectsCovered = dcount(ProjectId),
    AgentsCovered = dcount(AgentName),
    OwaspCategoriesDetected = make_set(OwaspCategory)
  by bin(TimeGenerated, 1d)
| order by TimeGenerated asc
```

### SOC 2 CC7.4 — Incident response

Pair Cognisafe alert data with your ITSM ticket records to evidence response timeliness:

```kql theme={null}
// Sentinel: CC7.4 evidence — mean time to respond by severity
CognisafeThreatEvents_CL
| where TimeGenerated > ago(90d)
| where Severity in ("critical", "high")
| join kind=leftouter (
    SecurityIncident
    | where ProviderName == "Cognisafe"
    | project IncidentCreationTime, RelatedAnalyticRuleId
  ) on $left.EventId == $right.RelatedAnalyticRuleId
| extend ResponseTimeHours = datetime_diff("hour", IncidentCreationTime, TimeGenerated)
| summarize AvgResponseHours = avg(ResponseTimeHours), P95ResponseHours = percentile(ResponseTimeHours, 95)
  by Severity
```

<Tip>
  Export the results of both queries as PDFs and store them in your GRC platform (Vanta, Drata, Secureframe, or equivalent) against the CC7.3 and CC7.4 controls. Run this monthly during audit preparation so you have a 12-month evidence trail before the auditor requests it.
</Tip>