feat(harfanglab): on-demand AV and YARA scan commands

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

integrations/harfanglab/scripts/av_scan.py

@@ -0,0 +1,47 @@
+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 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",
+    }
+    scan_type = str(inputs.get("scan_type") or "quick").lower()
+    if scan_type not in ("full", "quick", "resource"):
+        print(json.dumps({"error": "invalid scan_type: " + scan_type + " (expected full, quick or resource)"}))
+        sys.exit(1)
+    params = {"type": scan_type}
+    paths = csv(inputs.get("paths"))
+    if paths:
+        params["paths"] = paths
+    body = {
+        "targets": {"agents": [inputs.get("agent_id")]},
+        "actions": [{"value": "avScan", "params": params}],
+    }
+    print(json.dumps(request("POST", base + "/api/data/Job/", headers, body)))
+
+
+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)

integrations/harfanglab/scripts/yara_scan.py

@@ -0,0 +1,49 @@
+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 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",
+    }
+    params = {
+        "scanFilesystem": bool(inputs.get("scan_filesystem", True)),
+        "scanProcesses": bool(inputs.get("scan_processes", False)),
+    }
+    sources = csv(inputs.get("sources"))
+    if sources:
+        params["sources"] = sources
+    directories = csv(inputs.get("directories"))
+    if directories:
+        params["directoriesToScan"] = directories
+    body = {
+        "targets": {"agents": [inputs.get("agent_id")]},
+        "actions": [{"value": "yaraScan", "params": params}],
+    }
+    print(json.dumps(request("POST", base + "/api/data/Job/", headers, body)))
+
+
+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)