feat(harfanglab): HarfangLab EDR integration (71 analyst commands)

Endpoint management & isolation, policy assignment, IOC & whitelist (threat intelligence), security-event triage, telemetry hunting (processes, network, DNS, authentications, event logs, binaries), hash threat hunting, and forensic collection jobs (pipes, prefetch, run keys, scheduled tasks, drivers, services, processes, network, sessions, WMI, IOC scan, artifacts, RAM dump) plus their result-retrieval commands. Script-based (urllib, INTEGRATION_SECRETS/INPUTS contract; Authorization: Token header). Platform-specific XSOAR mirroring/fetch commands intentionally excluded.
2026-06-22 13:50:20 +02:00
parent e5237c0df6
commit 6b65c5f9df
72 changed files with 4313 additions and 0 deletions
@@ -0,0 +1,59 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def request(method, url, headers, body=None):
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=60) as resp:
+        raw = resp.read()
+        return json.loads(raw) if raw else {}
+
+
+def qs(params):
+    clean = {}
+    for k, v in params.items():
+        if v is None or v == "" or v == []:
+            continue
+        if isinstance(v, bool):
+            v = "true" if v else "false"
+        clean[k] = v
+    return ("?" + urllib.parse.urlencode(clean)) if clean else ""
+
+
+def csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+def main():
+    secrets = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    base = secrets.get("url", "").rstrip("/")
+    headers = {
+        "Authorization": "Token " + secrets.get("api_token", ""),
+        "Accept": "application/json",
+        "Content-Type": "application/json",
+    }
+    # === per-command logic ===
+    policy = inputs.get("policy")
+    agentid = inputs.get("agentid")
+    res = request("GET", base + "/api/data/endpoint/Policy/" + qs({"search": policy}), headers)
+    pid = None
+    for p in res.get("results", []):
+        if p.get("name") == policy:
+            pid = p.get("id")
+            break
+    if pid is not None:
+        out = request("POST", base + "/api/data/endpoint/Policy/" + str(pid) + "/add_agents/", headers, {"agent_ids": [agentid]})
+        print(json.dumps({"message": "Policy " + str(policy) + " assigned to agent " + str(agentid), "result": out}))
+    else:
+        print(json.dumps({"error": "Unknown policy " + str(policy)}))
+
+
+try:
+    main()
+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)