feat(elasticsearch): new Elasticsearch log-search integration

Elasticsearch REST API, 7 commands: search (DSL or query_string), count, get/
index document, list indices, delete by query. API-key auth, stdlib-only.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

integrations/elasticsearch/manifest.yaml

@@ -0,0 +1,94 @@
+id: elasticsearch
+name: Elasticsearch
+version: 1.0.0
+description: "Elasticsearch (REST API) — log search and enrichment for security data: run queries, count and read documents, list indices, index a document, and delete by query. API-key authentication; stdlib-only, no extra Python dependencies. Works with Elastic Security indices."
+changelog: "1.0.0 — Initial release: search, count, get document, index document, list indices, delete by query."
+category: siem
+
+# Per-instance configuration. Auth header 'Authorization: ApiKey <api_key>'.
+config_schema:
+  properties:
+    url:
+      type: string
+      description: "Elasticsearch base URL (e.g. https://es.example.com:9200)"
+    api_key:
+      type: string
+      description: "Elasticsearch API key (base64 id:key)"
+      x-soar-sensitive: true
+    insecure:
+      type: boolean
+      description: "Trust any TLS certificate (not secure)"
+      default: false
+  required:
+    - url
+    - api_key
+
+commands:
+  - id: search
+    name: elasticsearch-search
+    description: "Search an index. Provide a raw query DSL (query_json) or a simple query string."
+    risk: read
+    inputs_schema:
+      properties:
+        index: { type: string, description: "Index or index pattern (e.g. logs-*)" }
+        query_json: { type: string, description: "Raw Elasticsearch query DSL as a JSON object (advanced)" }
+        query_string: { type: string, description: "Simple query_string query (used if query_json is empty)" }
+        size: { type: number, description: "Max hits (default 50)" }
+      required: [index]
+    outputs_schema: { properties: {} }
+  - id: count
+    name: elasticsearch-count
+    description: "Count documents matching a query."
+    risk: read
+    inputs_schema:
+      properties:
+        index: { type: string, description: "Index or index pattern" }
+        query_string: { type: string, description: "Optional simple query_string (default all)" }
+      required: [index]
+    outputs_schema: { properties: {} }
+  - id: get_document
+    name: elasticsearch-get-document
+    description: "Get a single document by ID."
+    risk: read
+    inputs_schema:
+      properties:
+        index: { type: string, description: "Index name" }
+        doc_id: { type: string, description: "Document ID" }
+      required: [index, doc_id]
+    outputs_schema: { properties: {} }
+  - id: index_document
+    name: elasticsearch-index-document
+    description: "Index (create) a document."
+    inputs_schema:
+      properties:
+        index: { type: string, description: "Index name" }
+        document_json: { type: string, description: "Document body as a JSON object" }
+        doc_id: { type: string, description: "Optional document ID (auto-generated if omitted)" }
+      required: [index, document_json]
+    outputs_schema: { properties: {} }
+  - id: list_indices
+    name: elasticsearch-list-indices
+    description: "List indices."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }
+  - id: delete_by_query
+    name: elasticsearch-delete-by-query
+    description: "Delete documents matching a query."
+    inputs_schema:
+      properties:
+        index: { type: string, description: "Index name" }
+        query_json: { type: string, description: "Raw query DSL (JSON object) selecting documents to delete" }
+      required: [index, query_json]
+    outputs_schema: { properties: {} }
+
+  - id: test_connection
+    name: elasticsearch-test-connection
+    description: "Verify connectivity and the API key (used by the Test button)."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }

integrations/elasticsearch/scripts/count.py

@@ -0,0 +1,72 @@
+import json, os, sys, ssl, urllib.parse, urllib.request, urllib.error
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _ctx(cfg):
+    if cfg.get("insecure"):
+        c = ssl.create_default_context()
+        c.check_hostname = False
+        c.verify_mode = ssl.CERT_NONE
+        return c
+    return None
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("url", "")).rstrip("/") + path
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += "?" + urllib.parse.urlencode(clean)
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Authorization": "ApiKey " + str(cfg.get("api_key", "")), "Accept": "application/json"}
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _run(fn):
+    try:
+        print(json.dumps(fn(_cfg(), _inputs())))
+    except urllib.error.HTTPError as e:
+        print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
+        sys.exit(1)
+    except Exception as e:
+        print(json.dumps({"error": str(e)}))
+        sys.exit(1)
+
+
+q = lambda v: urllib.parse.quote(str(v), safe="")
+
+
+def _parse_json(s, field):
+    try:
+        return json.loads(s)
+    except Exception:
+        raise Exception(field + " must be a valid JSON object")
+
+
+def main(cfg, inputs):
+    index = inputs.get("index")
+    if not index:
+        raise Exception("index is required")
+    query_string = inputs.get("query_string")
+
+    if query_string:
+        body = {"query": {"query_string": {"query": query_string}}}
+    else:
+        body = {"query": {"match_all": {}}}
+
+    return request("POST", "/" + q(index) + "/_count", cfg, body=body)
+
+
+_run(main)

integrations/elasticsearch/scripts/delete_by_query.py

@@ -0,0 +1,75 @@
+import json, os, sys, ssl, urllib.parse, urllib.request, urllib.error
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _ctx(cfg):
+    if cfg.get("insecure"):
+        c = ssl.create_default_context()
+        c.check_hostname = False
+        c.verify_mode = ssl.CERT_NONE
+        return c
+    return None
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("url", "")).rstrip("/") + path
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += "?" + urllib.parse.urlencode(clean)
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Authorization": "ApiKey " + str(cfg.get("api_key", "")), "Accept": "application/json"}
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _run(fn):
+    try:
+        print(json.dumps(fn(_cfg(), _inputs())))
+    except urllib.error.HTTPError as e:
+        print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
+        sys.exit(1)
+    except Exception as e:
+        print(json.dumps({"error": str(e)}))
+        sys.exit(1)
+
+
+q = lambda v: urllib.parse.quote(str(v), safe="")
+
+
+def _parse_json(s, field):
+    try:
+        return json.loads(s)
+    except Exception:
+        raise Exception(field + " must be a valid JSON object")
+
+
+def main(cfg, inputs):
+    index = inputs.get("index")
+    if not index:
+        raise Exception("index is required")
+    query_json = inputs.get("query_json")
+    if not query_json:
+        raise Exception("query_json is required")
+    parsed = _parse_json(query_json, "query_json")
+
+    if "query" in parsed:
+        body = parsed
+    else:
+        body = {"query": parsed}
+
+    return request("POST", "/" + q(index) + "/_delete_by_query", cfg, body=body)
+
+
+_run(main)

integrations/elasticsearch/scripts/get_document.py

@@ -0,0 +1,69 @@
+import json, os, sys, ssl, urllib.parse, urllib.request, urllib.error
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _ctx(cfg):
+    if cfg.get("insecure"):
+        c = ssl.create_default_context()
+        c.check_hostname = False
+        c.verify_mode = ssl.CERT_NONE
+        return c
+    return None
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("url", "")).rstrip("/") + path
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += "?" + urllib.parse.urlencode(clean)
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Authorization": "ApiKey " + str(cfg.get("api_key", "")), "Accept": "application/json"}
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _run(fn):
+    try:
+        print(json.dumps(fn(_cfg(), _inputs())))
+    except urllib.error.HTTPError as e:
+        print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
+        sys.exit(1)
+    except Exception as e:
+        print(json.dumps({"error": str(e)}))
+        sys.exit(1)
+
+
+q = lambda v: urllib.parse.quote(str(v), safe="")
+
+
+def _parse_json(s, field):
+    try:
+        return json.loads(s)
+    except Exception:
+        raise Exception(field + " must be a valid JSON object")
+
+
+def main(cfg, inputs):
+    index = inputs.get("index")
+    if not index:
+        raise Exception("index is required")
+    doc_id = inputs.get("doc_id")
+    if not doc_id:
+        raise Exception("doc_id is required")
+
+    return request("GET", "/" + q(index) + "/_doc/" + q(doc_id), cfg)
+
+
+_run(main)

integrations/elasticsearch/scripts/index_document.py

@@ -0,0 +1,73 @@
+import json, os, sys, ssl, urllib.parse, urllib.request, urllib.error
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _ctx(cfg):
+    if cfg.get("insecure"):
+        c = ssl.create_default_context()
+        c.check_hostname = False
+        c.verify_mode = ssl.CERT_NONE
+        return c
+    return None
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("url", "")).rstrip("/") + path
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += "?" + urllib.parse.urlencode(clean)
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Authorization": "ApiKey " + str(cfg.get("api_key", "")), "Accept": "application/json"}
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _run(fn):
+    try:
+        print(json.dumps(fn(_cfg(), _inputs())))
+    except urllib.error.HTTPError as e:
+        print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
+        sys.exit(1)
+    except Exception as e:
+        print(json.dumps({"error": str(e)}))
+        sys.exit(1)
+
+
+q = lambda v: urllib.parse.quote(str(v), safe="")
+
+
+def _parse_json(s, field):
+    try:
+        return json.loads(s)
+    except Exception:
+        raise Exception(field + " must be a valid JSON object")
+
+
+def main(cfg, inputs):
+    index = inputs.get("index")
+    if not index:
+        raise Exception("index is required")
+    document_json = inputs.get("document_json")
+    if not document_json:
+        raise Exception("document_json is required")
+    doc = _parse_json(document_json, "document_json")
+    doc_id = inputs.get("doc_id")
+
+    if doc_id:
+        return request("PUT", "/" + q(index) + "/_doc/" + q(doc_id), cfg, body=doc)
+    return request("POST", "/" + q(index) + "/_doc", cfg, body=doc)
+
+
+_run(main)

integrations/elasticsearch/scripts/list_indices.py

@@ -0,0 +1,62 @@
+import json, os, sys, ssl, urllib.parse, urllib.request, urllib.error
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _ctx(cfg):
+    if cfg.get("insecure"):
+        c = ssl.create_default_context()
+        c.check_hostname = False
+        c.verify_mode = ssl.CERT_NONE
+        return c
+    return None
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("url", "")).rstrip("/") + path
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += "?" + urllib.parse.urlencode(clean)
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Authorization": "ApiKey " + str(cfg.get("api_key", "")), "Accept": "application/json"}
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _run(fn):
+    try:
+        print(json.dumps(fn(_cfg(), _inputs())))
+    except urllib.error.HTTPError as e:
+        print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
+        sys.exit(1)
+    except Exception as e:
+        print(json.dumps({"error": str(e)}))
+        sys.exit(1)
+
+
+q = lambda v: urllib.parse.quote(str(v), safe="")
+
+
+def _parse_json(s, field):
+    try:
+        return json.loads(s)
+    except Exception:
+        raise Exception(field + " must be a valid JSON object")
+
+
+def main(cfg, inputs):
+    return request("GET", "/_cat/indices", cfg, params={"format": "json"})
+
+
+_run(main)

integrations/elasticsearch/scripts/search.py

@@ -0,0 +1,81 @@
+import json, os, sys, ssl, urllib.parse, urllib.request, urllib.error
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _ctx(cfg):
+    if cfg.get("insecure"):
+        c = ssl.create_default_context()
+        c.check_hostname = False
+        c.verify_mode = ssl.CERT_NONE
+        return c
+    return None
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("url", "")).rstrip("/") + path
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += "?" + urllib.parse.urlencode(clean)
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Authorization": "ApiKey " + str(cfg.get("api_key", "")), "Accept": "application/json"}
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _run(fn):
+    try:
+        print(json.dumps(fn(_cfg(), _inputs())))
+    except urllib.error.HTTPError as e:
+        print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
+        sys.exit(1)
+    except Exception as e:
+        print(json.dumps({"error": str(e)}))
+        sys.exit(1)
+
+
+q = lambda v: urllib.parse.quote(str(v), safe="")
+
+
+def _parse_json(s, field):
+    try:
+        return json.loads(s)
+    except Exception:
+        raise Exception(field + " must be a valid JSON object")
+
+
+def main(cfg, inputs):
+    index = inputs.get("index")
+    if not index:
+        raise Exception("index is required")
+    query_json = inputs.get("query_json")
+    query_string = inputs.get("query_string")
+    size = inputs.get("size")
+
+    if query_json:
+        parsed = _parse_json(query_json, "query_json")
+        if "query" in parsed:
+            body = parsed
+            body.setdefault("size", int(size or 50))
+        else:
+            body = {"query": parsed, "size": int(size or 50)}
+    elif query_string:
+        body = {"query": {"query_string": {"query": query_string}}, "size": int(size or 50)}
+    else:
+        body = {"query": {"match_all": {}}, "size": int(size or 50)}
+
+    return request("POST", "/" + q(index) + "/_search", cfg, body=body)
+
+
+_run(main)

integrations/elasticsearch/scripts/test_connection.py

@@ -0,0 +1,63 @@
+import json, os, sys, ssl, urllib.parse, urllib.request, urllib.error
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _ctx(cfg):
+    if cfg.get("insecure"):
+        c = ssl.create_default_context()
+        c.check_hostname = False
+        c.verify_mode = ssl.CERT_NONE
+        return c
+    return None
+
+
+def request(method, path, cfg, body=None, params=None):
+    url = str(cfg.get("url", "")).rstrip("/") + path
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += "?" + urllib.parse.urlencode(clean)
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Authorization": "ApiKey " + str(cfg.get("api_key", "")), "Accept": "application/json"}
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _run(fn):
+    try:
+        print(json.dumps(fn(_cfg(), _inputs())))
+    except urllib.error.HTTPError as e:
+        print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
+        sys.exit(1)
+    except Exception as e:
+        print(json.dumps({"error": str(e)}))
+        sys.exit(1)
+
+
+q = lambda v: urllib.parse.quote(str(v), safe="")
+
+
+def _parse_json(s, field):
+    try:
+        return json.loads(s)
+    except Exception:
+        raise Exception(field + " must be a valid JSON object")
+
+
+def main(cfg, inputs):
+    request("GET", "/", cfg)
+    return {"ok": True}
+
+
+_run(main)