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>
91 lines
2.9 KiB
Python
91 lines
2.9 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():
|
|
q = (I.get("query") or "").strip()
|
|
if not (q.lower().startswith("search ") or q.startswith("|")):
|
|
q = "search " + q
|
|
data = {"search": q, "output_mode": "json"}
|
|
if I.get("earliest"):
|
|
data["earliest_time"] = I["earliest"]
|
|
if I.get("latest"):
|
|
data["latest_time"] = I["latest"]
|
|
path = "/servicesNS/nobody/" + qq(_app()) + "/search/jobs"
|
|
res = req("POST", path, data=data)
|
|
out({"sid": res.get("sid"), "result": 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)
|