Elastic/OpenSearch SIEM integration deep dive

Arbitex forwards audit events to Elasticsearch (and compatible OpenSearch deployments) via the Elasticsearch Bulk API using the ElasticConnector class (backend/app/services/siem/elastic.py). Events are serialized to OCSF v1.1 and delivered as NDJSON bulk requests to the configured index. The connector supports API key authentication, basic authentication, and Elastic Cloud ID for simplified endpoint configuration.

For a comparison of all seven Arbitex SIEM connectors, see the SIEM integration overview. For background on delivery path selection (Platform connector vs Outpost direct sink), see the SIEM integration guide.

---

Overview

The Elastic connector operates server-side on the Arbitex Platform. Events are buffered in memory and flushed in batches to the _bulk endpoint. Each bulk request contains action/source pairs in NDJSON format — one {"index":{"_index":"arbitex-ocsf"}} action line followed by one OCSF event document per event.

Architecture flow:

Arbitex Platform (event source)
        |
        v
  [In-memory buffer]  <-- async background flush task
        |         \
        |          \-- flush every ELASTIC_FLUSH_INTERVAL seconds
        v
  [Bulk NDJSON payload]
        |
        v
  POST {url}/_bulk
       Authorization: ApiKey <key>   (or Basic <credentials>)
       Content-Type: application/x-ndjson
        |
   -----+------
  |             |
HTTP 200     HTTP 4xx/5xx
  |             |
Check body   Retry or dead letter
"errors":true?
  |
  +-- per-item errors → dead letter
  |
  +-- all succeeded → acked

Important: The Elasticsearch Bulk API always returns HTTP 200, even when individual document indexing operations fail. The connector inspects the response body for "errors": true and routes failed items to the dead letter file on a per-document basis. Successfully indexed documents within the same batch are acknowledged normally.

Connector ID: elastic

Source file: backend/app/services/siem/elastic.py

Event format: OCSF v1.1.0, Elasticsearch Bulk API NDJSON

Target endpoint: POST {url}/_bulk

Content-Type: application/x-ndjson

---

Prerequisites

Elasticsearch / OpenSearch requirements:

• Elasticsearch 7.x+ or OpenSearch 1.x+ (Bulk API is stable across these versions)
• An API key or username/password with write access to the target index
• The target index or a matching index template must exist before events arrive
• Elastic Cloud deployments: Cloud ID is available in the Elastic Cloud console

Arbitex requirements:

• Platform version with ElasticConnector support
• At minimum one auth method: API key (ELASTIC_API_KEY) or basic auth (ELASTIC_USERNAME + ELASTIC_PASSWORD)
• At minimum one endpoint: direct URL (ELASTIC_URL) or Cloud ID (ELASTIC_CLOUD_ID)
• Write access to the dead letter directory (default /var/log/arbitex/)

Caution

Both a URL (or Cloud ID to decode to a URL) and at least one authentication method are required. The connector will refuse to start if neither ELASTIC_URL nor ELASTIC_CLOUD_ID is set, or if neither ELASTIC_API_KEY nor ELASTIC_USERNAME/ELASTIC_PASSWORD are set.

---

Step-by-step setup

1. Create a target index or index template

   Before configuring the connector, create the arbitex-ocsf index or an index template that matches arbitex-ocsf-*. See the index template section for the recommended mapping.

   Create index directly

   curl -X PUT "https://elastic.example.com:9200/arbitex-ocsf" \
     -H "Authorization: ApiKey <key>" \
     -H "Content-Type: application/json" \
     -d '{
       "settings": {
         "number_of_shards": 1,
         "number_of_replicas": 1
       }
     }'

   Apply index template

   See the index template section for the full template definition. Apply it with:

   curl -X PUT "https://elastic.example.com:9200/_index_template/arbitex-ocsf-template" \
     -H "Authorization: ApiKey <key>" \
     -H "Content-Type: application/json" \
     -d @arbitex-ocsf-template.json

2. Create an API key

   Elasticsearch API key

   curl -X POST "https://elastic.example.com:9200/_security/api_key" \
     -H "Authorization: Basic <admin-credentials>" \
     -H "Content-Type: application/json" \
     -d '{
       "name": "arbitex-platform",
       "role_descriptors": {
         "arbitex-writer": {
           "cluster": ["monitor"],
           "indices": [
             {
               "names": ["arbitex-ocsf*"],
               "privileges": ["create_index", "index", "write"]
             }
           ]
         }
       }
     }'

   The response includes an encoded field — this is your ELASTIC_API_KEY value.

   Kibana UI

   In Kibana, navigate to Stack Management > API Keys > Create API key.

   • Name: arbitex-platform
   • Control security privileges: restrict to index and create_index on arbitex-ocsf*

   Copy the encoded key value shown after creation.

3. Determine your endpoint

   Direct URL

   Use the Elasticsearch HTTPS endpoint directly:

   ELASTIC_URL=https://elastic.example.com:9200

   For Elastic Cloud deployments, you can find this in the Cloud console under your deployment > Endpoints > Elasticsearch.

   Cloud ID

   Use the Cloud ID from the Elastic Cloud console. The connector decodes the Cloud ID to extract the Elasticsearch HTTPS URL automatically.

   The Cloud ID format is deployment-name:base64(es_host$kibana_host). Set:

   ELASTIC_CLOUD_ID=my-deployment:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyRlc19ob3N0JGtpYmFuYV9ob3N0

   You do not need to set ELASTIC_URL when using ELASTIC_CLOUD_ID.

4. Set environment variables

   API key auth (direct URL)

   ELASTIC_URL=https://elastic.example.com:9200
   ELASTIC_API_KEY=ABA1FA15E000-example-encoded-api-key
   ELASTIC_INDEX=arbitex-ocsf
   ELASTIC_BATCH_SIZE=100
   ELASTIC_FLUSH_INTERVAL=5
   ELASTIC_MAX_RETRIES=3
   ELASTIC_DEAD_LETTER_PATH=/var/log/arbitex/elastic_dead_letter.jsonl

   API key auth (Cloud ID)

   ELASTIC_CLOUD_ID=my-deployment:dXMtZWFzdC0xLmF3cy5mb3VuZC5pbyRlc19ob3N0JGtpYmFuYV9ob3N0
   ELASTIC_API_KEY=ABA1FA15E000-example-encoded-api-key
   ELASTIC_INDEX=arbitex-ocsf
   ELASTIC_BATCH_SIZE=100
   ELASTIC_FLUSH_INTERVAL=5
   ELASTIC_MAX_RETRIES=3
   ELASTIC_DEAD_LETTER_PATH=/var/log/arbitex/elastic_dead_letter.jsonl

   Basic auth

   ELASTIC_URL=https://elastic.example.com:9200
   ELASTIC_USERNAME=arbitex-service-user
   ELASTIC_PASSWORD=your_password
   ELASTIC_INDEX=arbitex-ocsf
   ELASTIC_BATCH_SIZE=100
   ELASTIC_FLUSH_INTERVAL=5
   ELASTIC_MAX_RETRIES=3
   ELASTIC_DEAD_LETTER_PATH=/var/log/arbitex/elastic_dead_letter.jsonl

   Kubernetes secret

   apiVersion: v1
   kind: Secret
   metadata:
     name: arbitex-siem-elastic
     namespace: arbitex
   type: Opaque
   stringData:
     ELASTIC_URL: "https://elastic.example.com:9200"
     ELASTIC_API_KEY: "your-encoded-api-key"
     ELASTIC_INDEX: "arbitex-ocsf"
     ELASTIC_BATCH_SIZE: "100"
     ELASTIC_FLUSH_INTERVAL: "5"
     ELASTIC_MAX_RETRIES: "3"
     ELASTIC_DEAD_LETTER_PATH: "/var/log/arbitex/elastic_dead_letter.jsonl"

5. Verify the dead letter directory

   mkdir -p /var/log/arbitex
   chown arbitex:arbitex /var/log/arbitex
   chmod 750 /var/log/arbitex

6. Restart the Arbitex Platform

   # Kubernetes
   kubectl rollout restart deployment/arbitex-platform -n arbitex
   
   # Docker Compose
   docker compose up -d --force-recreate platform

7. Confirm event delivery

   Check the cluster health and verify documents are arriving:

   # Cluster health
   curl -H "Authorization: ApiKey <key>" \
     https://elastic.example.com:9200/_cluster/health
   
   # Count documents in the index
   curl -H "Authorization: ApiKey <key>" \
     https://elastic.example.com:9200/arbitex-ocsf/_count

   The connector’s health check uses GET /_cluster/health. HEALTHY maps to green or yellow cluster status. DEGRADED maps to red. ERROR maps to connection failure.

---

Configuration reference

Environment variable	Required	Default	Description
ELASTIC_URL	Yes (or Cloud ID)	—	Elasticsearch HTTPS endpoint, e.g. https://elastic.example.com:9200
ELASTIC_API_KEY	Yes (or user+pass)	—	Encoded Elasticsearch API key for Authorization: ApiKey header
ELASTIC_INDEX	No	arbitex-ocsf	Target index name for all Arbitex events
ELASTIC_CLOUD_ID	No	—	Elastic Cloud ID (alternative to ELASTIC_URL; connector decodes to extract ES URL)
ELASTIC_USERNAME	No	—	Basic auth username (used if ELASTIC_API_KEY is not set)
ELASTIC_PASSWORD	No	—	Basic auth password (used with ELASTIC_USERNAME)
ELASTIC_BATCH_SIZE	No	100	Maximum number of events per bulk request
ELASTIC_FLUSH_INTERVAL	No	5	Seconds between background flush cycles
ELASTIC_MAX_RETRIES	No	3	Maximum retry attempts for retriable errors
ELASTIC_DEAD_LETTER_PATH	No	/var/log/arbitex/elastic_dead_letter.jsonl	Path to the JSONL dead letter file

Authentication precedence: API key takes precedence over basic auth. If ELASTIC_API_KEY is set, the Authorization: ApiKey <key> header is used. If only ELASTIC_USERNAME and ELASTIC_PASSWORD are set, httpx.BasicAuth is used.

Endpoint precedence: If ELASTIC_CLOUD_ID is set, the connector decodes it to extract the Elasticsearch HTTPS URL. If ELASTIC_URL is also set, ELASTIC_URL takes precedence.

---

Sample configuration

Bulk API payload structure

Each flush generates an NDJSON bulk request. The payload alternates action lines and document lines:

{"index":{"_index":"arbitex-ocsf"}}
{"class_uid":2001,"severity_id":4,"actor":{"user":{"uid":"[email protected]","org_uid":"org_abc123"}},"finding_info":{"title":"Credential detected in response"},"raw_data":"credint_hit","time":1741824000123}
{"index":{"_index":"arbitex-ocsf"}}
{"class_uid":6003,"actor":{"user":{"uid":"[email protected]"}},"unmapped":{"model_id":"gpt-4o","latency_ms":342,"provider":"openai"},"time":1741824001456}
{"index":{"_index":"arbitex-ocsf"}}
{"class_uid":3002,"status":"Failure","actor":{"user":{"uid":"[email protected]"}},"src_endpoint":{"ip":"198.51.100.42"},"time":1741824002789}

The _index value in each action line is always the value of ELASTIC_INDEX.

Bulk API response handling

The connector checks the response body for partial failures:

{
  "took": 12,
  "errors": true,
  "items": [
    {"index": {"_index": "arbitex-ocsf", "_id": "abc", "result": "created", "status": 201}},
    {"index": {"_index": "arbitex-ocsf", "_id": "def", "error": {"type": "mapper_parsing_exception", "reason": "..."}, "status": 400}},
    {"index": {"_index": "arbitex-ocsf", "_id": "ghi", "result": "created", "status": 201}}
  ]
}

When "errors": true, the connector iterates through items and writes any item with a non-2xx status to the dead letter file. Successfully indexed items are counted as delivered.

Dead letter file format

{"event":{"class_uid":2001,"severity_id":4},"error":"mapper_parsing_exception: failed to parse field [class_uid]","connector":"elastic","timestamp":1741824005.0}

---

Index template

Apply this index template before deploying the connector to ensure correct field types for all OCSF fields used in queries. Without explicit mappings, Elasticsearch may auto-map fields incorrectly (e.g., mapping class_uid as long with a coercion error, or mapping unmapped.latency_ms as a string if the first ingested value is a string).

{
  "index_patterns": ["arbitex-ocsf-*", "arbitex-ocsf"],
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 1,
      "index.codec": "best_compression"
    },
    "mappings": {
      "dynamic": true,
      "properties": {
        "class_uid": {
          "type": "integer"
        },
        "severity_id": {
          "type": "integer"
        },
        "time": {
          "type": "date",
          "format": "epoch_millis"
        },
        "status": {
          "type": "keyword"
        },
        "raw_data": {
          "type": "keyword"
        },
        "message": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        },
        "actor": {
          "properties": {
            "user": {
              "properties": {
                "uid": {
                  "type": "keyword"
                },
                "org_uid": {
                  "type": "keyword"
                }
              }
            }
          }
        },
        "src_endpoint": {
          "properties": {
            "ip": {
              "type": "ip"
            }
          }
        },
        "finding_info": {
          "properties": {
            "title": {
              "type": "keyword"
            }
          }
        },
        "unmapped": {
          "properties": {
            "model_id": {
              "type": "keyword"
            },
            "provider": {
              "type": "keyword"
            },
            "latency_ms": {
              "type": "float"
            },
            "category": {
              "type": "keyword"
            },
            "action": {
              "type": "keyword"
            }
          }
        }
      }
    }
  },
  "priority": 100,
  "version": 1,
  "_meta": {
    "description": "Arbitex OCSF v1.1 event index template"
  }
}

Save this as arbitex-ocsf-template.json and apply with:

curl -X PUT "https://elastic.example.com:9200/_index_template/arbitex-ocsf-template" \
  -H "Authorization: ApiKey <key>" \
  -H "Content-Type: application/json" \
  -d @arbitex-ocsf-template.json

Tip

Apply the index template before the connector creates the index for the first time. If the index already exists with incorrect mappings, you will need to reindex existing data after updating the template, or create a new index with an alias.

---

Example queries

The following queries use the Elasticsearch Query DSL (JSON). They can be run via the _search API directly, through Kibana Dev Tools, or used as the basis for Kibana dashboards and Elastic SIEM detection rules.

DLP blocks by category and action

Returns DLP enforcement events with severity High or Critical (severity_id 4 or 5), aggregated by category and action:

POST /arbitex-ocsf/_search
{
  "query": {
    "bool": {
      "must": [
        {"term": {"class_uid": 2001}},
        {"range": {"severity_id": {"gte": 4}}}
      ]
    }
  },
  "aggs": {
    "by_category": {
      "terms": {"field": "unmapped.category"},
      "aggs": {
        "by_action": {
          "terms": {"field": "unmapped.action"}
        }
      }
    }
  },
  "size": 0
}

Authentication failures by user and source IP

Identifies users with repeated authentication failures:

POST /arbitex-ocsf/_search
{
  "query": {
    "bool": {
      "must": [
        {"term": {"class_uid": 3002}},
        {"term": {"status": "Failure"}}
      ]
    }
  },
  "aggs": {
    "by_user": {
      "terms": {"field": "actor.user.uid"},
      "aggs": {
        "by_ip": {
          "terms": {"field": "src_endpoint.ip"}
        }
      }
    }
  },
  "size": 0
}

Credential detections

Lists all events where the CredInt scanner flagged credentials in a request or response:

POST /arbitex-ocsf/_search
{
  "query": {
    "bool": {
      "must": [
        {"term": {"class_uid": 2001}},
        {"term": {"raw_data": "credint_hit"}}
      ]
    }
  },
  "sort": [{"time": {"order": "desc"}}],
  "_source": ["time", "actor.user.uid", "finding_info.title", "actor.user.org_uid"],
  "size": 50
}

High-latency API calls with average by model

Returns API activity events with latency above 5,000 ms, with per-model average latency aggregation:

POST /arbitex-ocsf/_search
{
  "query": {
    "bool": {
      "must": [
        {"term": {"class_uid": 6003}},
        {"range": {"unmapped.latency_ms": {"gt": 5000}}}
      ]
    }
  },
  "aggs": {
    "by_model": {
      "terms": {"field": "unmapped.model_id.keyword"},
      "aggs": {
        "avg_latency": {
          "avg": {"field": "unmapped.latency_ms"}
        }
      }
    }
  },
  "size": 0
}

Admin configuration changes

Returns all account change events (config changes, user creation, API key revocations), sorted by most recent:

POST /arbitex-ocsf/_search
{
  "query": {
    "term": {"class_uid": 3004}
  },
  "sort": [{"time": {"order": "desc"}}],
  "_source": ["time", "actor.user.uid", "raw_data", "message"],
  "size": 100
}

DLP event count over time (date histogram)

Suitable for Kibana dashboards showing DLP enforcement volume by hour:

POST /arbitex-ocsf/_search
{
  "query": {
    "bool": {
      "must": [
        {"term": {"class_uid": 2001}},
        {"range": {"time": {"gte": "now-24h"}}}
      ]
    }
  },
  "aggs": {
    "events_over_time": {
      "date_histogram": {
        "field": "time",
        "calendar_interval": "1h"
      },
      "aggs": {
        "by_action": {
          "terms": {"field": "unmapped.action"}
        }
      }
    }
  },
  "size": 0
}

---

Troubleshooting

Events not appearing in Elasticsearch

Symptom: No documents appear in the arbitex-ocsf index after platform startup and triggering test actions.

Diagnosis:

1. Check platform logs for connector errors: grep "elastic" /var/log/arbitex/platform.log
2. Verify cluster health: curl -H "Authorization: ApiKey <key>" https://elastic.example.com:9200/_cluster/health
3. Check if the dead letter file is growing: wc -l /var/log/arbitex/elastic_dead_letter.jsonl

Resolution:

• If cluster health returns red, resolve the cluster issue before events will index reliably.
• If the dead letter file contains "HTTP 401" or "HTTP 403" errors, the API key or credentials are invalid. Create a new API key with write permissions to arbitex-ocsf*.
• If no errors appear in logs or dead letter, confirm the flush interval has elapsed (ELASTIC_FLUSH_INTERVAL seconds) and that at least one event has been generated.

Bulk API returns 200 but “errors”: true

Symptom: Platform logs show successful HTTP 200 responses, but some events appear in the dead letter file with mapper_parsing_exception errors.

Diagnosis: The index has incorrect or missing mappings for one or more OCSF fields.

Resolution:

1. Apply the index template to establish correct field mappings.
2. If the index already exists with incorrect dynamic mappings, delete and recreate it (or create a new index and use an alias).
3. Examine the dead letter entries to identify which specific fields are causing mapping exceptions.

Cloud ID decoding fails

Symptom: Platform logs show an error parsing the Cloud ID, and the connector fails to start.

Diagnosis: The Cloud ID format is deployment-name:base64_encoded_value. The base64 portion must decode to es_host$kibana_host. Verify the Cloud ID is copied correctly from the Elastic Cloud console — it should not contain spaces or newlines.

Resolution: Copy the Cloud ID directly from Elastic Cloud > Deployment > Manage > Copy Cloud ID. Alternatively, decode the endpoint URL directly from the Cloud console and set ELASTIC_URL instead.

Authentication confusion (API key vs basic auth)

Symptom: Events are successfully delivered using basic auth, but after setting ELASTIC_API_KEY, delivery starts failing.

Diagnosis: API key takes precedence over basic auth. If both are set and the API key is invalid, the connector will use the invalid API key rather than falling back to basic auth.

Resolution: Use only one authentication method. Remove or unset the other to avoid ambiguity. Verify the API key is valid:

curl -H "Authorization: ApiKey <key>" \
  https://elastic.example.com:9200/_security/authenticate

Dead letter file growing with mapper exceptions

Symptom: Dead letter file contains many entries with "mapper_parsing_exception" for class_uid, severity_id, or unmapped.latency_ms.

Explanation: Without the index template, Elasticsearch dynamic mapping may incorrectly type these fields (e.g., unmapped.latency_ms as long when the first document has an integer value, then failing when a float arrives).

Resolution:

1. Apply the index template from the index template section.
2. Reindex or delete and recreate the index to apply correct mappings.
3. Replay dead letter events after correcting the mapping:

python3 - <<'EOF'
import json, urllib.request, os

url = os.environ["ELASTIC_URL"].rstrip("/")
key = os.environ["ELASTIC_API_KEY"]
index = os.environ.get("ELASTIC_INDEX", "arbitex-ocsf")

with open(os.environ.get("ELASTIC_DEAD_LETTER_PATH",
          "/var/log/arbitex/elastic_dead_letter.jsonl")) as f:
    for line in f:
        record = json.loads(line)
        action = json.dumps({"index": {"_index": index}})
        event = json.dumps(record["event"])
        payload = (action + "\n" + event + "\n").encode()
        req = urllib.request.Request(f"{url}/_bulk", data=payload,
            headers={"Authorization": f"ApiKey {key}",
                     "Content-Type": "application/x-ndjson"})
        with urllib.request.urlopen(req) as resp:
            body = json.loads(resp.read())
            if body.get("errors"):
                print("PARTIAL FAILURE:", body["items"])
            else:
                print("OK: replayed", len(body["items"]), "events")
EOF

Cluster health is red

Symptom: The connector health check reports DEGRADED or ERROR. /_cluster/health returns "status": "red".

Explanation: A red cluster status means one or more primary shards are unassigned. The connector maps red cluster status to DEGRADED. Connection failure maps to ERROR.

Resolution: Resolve the underlying cluster issue (unassigned shards, node failure) before expecting reliable event delivery. Events continue to buffer in memory during the DEGRADED state, but the buffer is bounded — events exceeding the buffer will go to the dead letter file.