Guillaume BOURGEOIS
dd1ca83d7d
20 commands over the management port (8089) with token or basic auth: SPL search (oneshot) as an ingestion source with an OCSF mapper for notable/CIM findings; async search jobs (create/status/results); index listing; event submission (receivers/simple + HEC); the complete KV Store command set (collection create/config/delete, list, data list/add/delete, entry search/delete/update); user list/delete; and a connectivity test. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
import json, os, ssl, sys, base64, urllib.parse, urllib.request, urllib.error
|
|
|
|
S = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
|
I = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
|
|
|
|
|
def qq(v):
|
|
return urllib.parse.quote(str(v), safe="")
|
|
|
|
|
|
def _base():
|
|
host = (S.get("host") or "").strip()
|
|
if "://" in host:
|
|
host = host.split("://", 1)[1]
|
|
host = host.strip("/")
|
|
if ":" not in host:
|
|
host = host + ":" + str(S.get("port") or "8089").strip()
|
|
return "https://" + host
|
|
|
|
|
|
def _auth():
|
|
if str(S.get("auth_type") or "token").lower() == "basic":
|
|
raw = (S.get("username") or "") + ":" + (S.get("password") or "")
|
|
return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("ascii")
|
|
return "Bearer " + (S.get("password") or "")
|
|
|
|
|
|
def _ctx():
|
|
v = S.get("verify")
|
|
if isinstance(v, str):
|
|
v = v.lower() in ("1", "true", "yes")
|
|
return ssl.create_default_context() if v else ssl._create_unverified_context()
|
|
|
|
|
|
def _app(default="search"):
|
|
return I.get("app") or I.get("app_name") or S.get("app") or default
|
|
|
|
|
|
def req(method, path, params=None, data=None, base=None, auth=None, ctype=None):
|
|
url = (base or _base()) + path
|
|
if params:
|
|
url += "?" + urllib.parse.urlencode(params, doseq=True)
|
|
headers = {"Authorization": auth or _auth(), "Accept": "application/json"}
|
|
body = None
|
|
if data is not None:
|
|
if ctype == "json":
|
|
body = json.dumps(data).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
elif isinstance(data, dict):
|
|
body = urllib.parse.urlencode(data, doseq=True).encode("utf-8")
|
|
headers["Content-Type"] = "application/x-www-form-urlencoded"
|
|
else:
|
|
body = data.encode("utf-8") if isinstance(data, str) else data
|
|
headers["Content-Type"] = ctype or "text/plain"
|
|
request = urllib.request.Request(url, data=body, headers=headers, method=method)
|
|
with urllib.request.urlopen(request, timeout=120, context=_ctx()) as resp:
|
|
raw = resp.read()
|
|
try:
|
|
return json.loads(raw) if raw else {}
|
|
except ValueError:
|
|
return {"raw": raw.decode("utf-8", "replace")}
|
|
|
|
|
|
def out(value):
|
|
print(json.dumps(value))
|
|
|
|
|
|
|
|
def run():
|
|
hec_url = (S.get("hec_url") or "").rstrip("/")
|
|
hec_token = S.get("hec_token") or ""
|
|
if not hec_url or not hec_token:
|
|
out({"error": "hec_url and hec_token must be configured on the instance"})
|
|
return
|
|
ev = I.get("event")
|
|
if isinstance(ev, str):
|
|
try:
|
|
ev = json.loads(ev)
|
|
except ValueError:
|
|
pass
|
|
payload = {"event": ev}
|
|
for k in ("index", "sourcetype", "source", "host"):
|
|
if I.get(k):
|
|
payload[k] = I[k]
|
|
if I.get("fields"):
|
|
payload["fields"] = I["fields"]
|
|
res = req("POST", "/services/collector/event", base=hec_url,
|
|
auth="Splunk " + hec_token, data=payload, ctype="json")
|
|
out(res)
|
|
|
|
|
|
try:
|
|
run()
|
|
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)
|