feat(rapid7-insightidr): InsightIDR integration (19 commands + OCSF ingestion)

REST API integration for Rapid7 InsightIDR. Investigation ingestion
(list_investigations) with an exhaustive OCSF mapper and a bundled default
incident type, plus 18 commands across investigations (list/get/search/
create/update/assign/set-status/bulk-close), investigation alerts and Rapid7
product alerts, custom threat indicators (add/replace), log management and
LEQL log/log-set queries with downloads, and user directory search.

API v1/v2 selectable per instance (is_v2) and per command (api_version);
multi-customer query parameter supported on v2 calls.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

integrations/rapid7-insightidr/scripts/add_threat_indicators.py

@@ -0,0 +1,73 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    return base, headers
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _list(v):
+    if v in (None, ""):
+        return []
+    if isinstance(v, list):
+        return [str(x).strip() for x in v if str(x).strip()]
+    return [p.strip() for p in str(v).split(",") if p.strip()]
+
+
+def _prune(d):
+    return {k: v for k, v in d.items() if v}
+
+
+def request(method, path, params=None, body=None):
+    base, headers = _cfg()
+    url = base + path.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
+    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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+# Endpoint suffix differs between add and replace; this script adds.
+ACTION = "add"
+
+
+def run():
+    inp = _inputs()
+    body = _prune({
+        "ips": _list(inp.get("ip_addresses")),
+        "hashes": _list(inp.get("hashes")),
+        "domain_names": _list(inp.get("domain_names")),
+        "urls": _list(inp.get("url")),
+    })
+    results = []
+    for key in _list(inp.get("key")):
+        out = request(
+            "POST",
+            "idr/v1/customthreats/key/" + urllib.parse.quote(key, safe="") + "/indicators/" + ACTION,
+            params={"format": "json"},
+            body=body,
+        )
+        results.append(out.get("threat", out))
+    print(json.dumps({"data": results}))
+
+
+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)

integrations/rapid7-insightidr/scripts/assign_user.py

@@ -0,0 +1,69 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    multi = str(s.get("is_multi_customer") or "").strip().lower() in ("1", "true", "yes")
+    is_v2 = str(s.get("is_v2") or "true").strip().lower() in ("1", "true", "yes")
+    return base, headers, multi, is_v2
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _api_version(inp, is_v2):
+    v = str(inp.get("api_version") or "Default").strip()
+    return v if v in ("V1", "V2") else ("V2" if is_v2 else "V1")
+
+
+def _list(v):
+    if v in (None, ""):
+        return []
+    if isinstance(v, list):
+        return [str(x).strip() for x in v if str(x).strip()]
+    return [p.strip() for p in str(v).split(",") if p.strip()]
+
+
+def request(method, path, params=None, body=None):
+    base, headers, _, _ = _cfg()
+    url = base + path.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
+    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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    _, _, multi, is_v2 = _cfg()
+    inp = _inputs()
+    api_version = _api_version(inp, is_v2)
+    email = inp.get("user_email_address")
+    params = {"multi-customer": "true" if (multi and api_version == "V2") else None}
+    results = []
+    for inv_id in _list(inp.get("investigation_id")):
+        results.append(request(
+            "PUT",
+            "idr/" + api_version.lower() + "/investigations/" + urllib.parse.quote(inv_id, safe="") + "/assignee",
+            params=params,
+            body={"user_email_address": email},
+        ))
+    print(json.dumps({"data": results}))
+
+
+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)

integrations/rapid7-insightidr/scripts/close_investigations.py

@@ -0,0 +1,54 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    return base, headers
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _prune(d):
+    return {k: v for k, v in d.items() if v not in (None, "", {}, [])}
+
+
+def request(method, path, body=None):
+    base, headers = _cfg()
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    req = urllib.request.Request(base + path.lstrip("/"), data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    inp = _inputs()
+    disposition = inp.get("disposition") or "Not Applicable"
+    max_close = inp.get("max_investigations_to_close")
+    body = _prune({
+        "source": inp.get("source"),
+        "alert_type": inp.get("alert_type"),
+        "disposition": disposition.replace(" ", "_"),
+        "detection_rule_rrn": inp.get("detection_rule_rrn"),
+        "from": inp.get("start_time"),
+        "to": inp.get("end_time"),
+        "max_investigations_to_close": int(max_close) if max_close not in (None, "") else None,
+    })
+    out = request("POST", "idr/v2/investigations/bulk_close", body=body)
+    ids = out.get("ids", [])
+    print(json.dumps({"ids": ids, "data": [{"id": i, "status": "CLOSED"} for i in ids]}))
+
+
+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)

integrations/rapid7-insightidr/scripts/create_investigation.py

@@ -0,0 +1,49 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    return base, headers
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _prune(d):
+    return {k: v for k, v in d.items() if v not in (None, "", {}, [])}
+
+
+def request(method, path, body=None):
+    base, headers = _cfg()
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    req = urllib.request.Request(base + path.lstrip("/"), data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    inp = _inputs()
+    disposition = str(inp.get("disposition") or "Undecided").replace(" ", "_")
+    body = _prune({
+        "assignee": _prune({"email": inp.get("user_email_address")}),
+        "disposition": disposition,
+        "priority": inp.get("priority") or "Unspecified",
+        "status": inp.get("status") or "Open",
+        "title": inp.get("title"),
+    })
+    print(json.dumps(request("POST", "idr/v2/investigations", body=body)))
+
+
+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)

integrations/rapid7-insightidr/scripts/download_logs.py

@@ -0,0 +1,52 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Accept": "application/json"}
+    return base, headers
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def run():
+    base, headers = _cfg()
+    inp = _inputs()
+    start_time = inp.get("start_time")
+    end_time = inp.get("end_time")
+    time_range = inp.get("time_range")
+    if not (start_time or end_time or time_range):
+        time_range = "Last 3 days"
+    params = {
+        "from": start_time,
+        "to": end_time,
+        "time_range": time_range,
+        "query": inp.get("query"),
+        "limit": inp.get("limit"),
+    }
+    clean = {k: v for k, v in params.items() if v not in (None, "")}
+    # Up to 10 log IDs are joined with ':' in the path.
+    log_ids = str(inp.get("log_ids", "")).replace(",", ":")
+    url = base + "log_search/download/logs/" + urllib.parse.quote(log_ids, safe=":")
+    if clean:
+        url += "?" + urllib.parse.urlencode(clean, doseq=True)
+    dl_headers = dict(headers)
+    dl_headers["Accept-Encoding"] = ""
+    req = urllib.request.Request(url, headers=dl_headers, method="GET")
+    with urllib.request.urlopen(req, timeout=120) as r:
+        content = r.read().decode("utf-8", "replace")
+    print(json.dumps({"content": content, "log_ids": inp.get("log_ids")}))
+
+
+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)

integrations/rapid7-insightidr/scripts/get_investigation.py

@@ -0,0 +1,60 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    multi = str(s.get("is_multi_customer") or "").strip().lower() in ("1", "true", "yes")
+    is_v2 = str(s.get("is_v2") or "true").strip().lower() in ("1", "true", "yes")
+    return base, headers, multi, is_v2
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _api_version(inp, is_v2):
+    v = str(inp.get("api_version") or "Default").strip()
+    return v if v in ("V1", "V2") else ("V2" if is_v2 else "V1")
+
+
+def request(method, path, params=None, body=None):
+    base, headers, _, _ = _cfg()
+    url = base + path.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
+    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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    _, _, multi, is_v2 = _cfg()
+    inp = _inputs()
+    api_version = _api_version(inp, is_v2)
+    inv_id = inp.get("investigation_id", "")
+    if api_version == "V2":
+        out = request("GET", "idr/v2/investigations/" + urllib.parse.quote(inv_id, safe=""),
+                      params={"multi-customer": "true" if multi else None})
+        print(json.dumps(out if out.get("rrn") else {}))
+        return
+    # V1 has no get-by-id endpoint: list and match on id.
+    data = request("GET", "idr/v1/investigations", params={"size": 1000}).get("data", [])
+    match = next((i for i in data if i.get("id") == inv_id), {})
+    print(json.dumps(match))
+
+
+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)

integrations/rapid7-insightidr/scripts/list_investigation_alerts.py

@@ -0,0 +1,51 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    multi = str(s.get("is_multi_customer") or "").strip().lower() in ("1", "true", "yes")
+    return base, headers, multi
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def request(method, path, params=None):
+    base, headers, _ = _cfg()
+    url = base + path.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
+    req = urllib.request.Request(url, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    _, _, multi = _cfg()
+    inp = _inputs()
+    inv_id = inp.get("investigation_id", "")
+    out = request("GET", "idr/v2/investigations/" + urllib.parse.quote(inv_id, safe="") + "/alerts",
+                  params={"multi-customer": "true" if multi else None})
+    data = out.get("data", [])
+    all_results = str(inp.get("all_results") or "false").strip().lower() in ("1", "true", "yes")
+    if not all_results:
+        limit = int(inp.get("limit") or 50)
+        data = data[:limit]
+    print(json.dumps({"rrn": inv_id, "alert": data}))
+
+
+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)

integrations/rapid7-insightidr/scripts/list_investigation_product_alerts.py

@@ -0,0 +1,65 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    multi = str(s.get("is_multi_customer") or "").strip().lower() in ("1", "true", "yes")
+    return base, headers, multi
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def request(method, path, params=None):
+    base, headers, _ = _cfg()
+    url = base + path.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
+    req = urllib.request.Request(url, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _flatten(response):
+    # Each entry holds a product 'type' plus one or more product alert lists/dicts.
+    rows = []
+    for result in response or []:
+        for product_name in list(result.keys()):
+            value = result[product_name]
+            if isinstance(value, list):
+                for alert in value:
+                    rows.append(dict(alert, name=result.get("type")))
+            elif isinstance(value, dict):
+                rows.append(dict(value, name=result.get("type")))
+    return rows
+
+
+def run():
+    _, _, multi = _cfg()
+    inp = _inputs()
+    inv_id = inp.get("investigation_id", "")
+    out = request("GET", "idr/v2/investigations/" + urllib.parse.quote(inv_id, safe="") + "/rapid7-product-alerts",
+                  params={"multi-customer": "true" if multi else None})
+    data = _flatten(out if isinstance(out, list) else out.get("data", []))
+    all_results = str(inp.get("all_results") or "false").strip().lower() in ("1", "true", "yes")
+    if not all_results:
+        limit = int(inp.get("limit") or 50)
+        data = data[:limit]
+    print(json.dumps({"rrn": inv_id, "ProductAlert": data}))
+
+
+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)

integrations/rapid7-insightidr/scripts/list_investigations.py

@@ -0,0 +1,93 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+from datetime import datetime, timedelta, timezone
+
+ISO = "%Y-%m-%dT%H:%M:%SZ"
+_UNITS = {"second": 1, "minute": 60, "hour": 3600, "day": 86400, "week": 604800, "month": 2592000, "year": 31536000}
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    multi = str(s.get("is_multi_customer") or "").strip().lower() in ("1", "true", "yes")
+    is_v2 = str(s.get("is_v2") or "true").strip().lower() in ("1", "true", "yes")
+    return base, headers, multi, is_v2
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _api_version(inp, is_v2):
+    v = str(inp.get("api_version") or "Default").strip()
+    return v if v in ("V1", "V2") else ("V2" if is_v2 else "V1")
+
+
+def _rel_seconds(text):
+    if not text:
+        return None
+    num = unit = None
+    for t in str(text).lower().replace("last", "").split():
+        if t.isdigit():
+            num = int(t)
+        elif t.rstrip("s") in _UNITS:
+            unit = t.rstrip("s")
+    return num * _UNITS[unit] if (num is not None and unit) else None
+
+
+def request(method, path, params=None, body=None):
+    base, headers, _, _ = _cfg()
+    url = base + path.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
+    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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    _, _, multi, is_v2 = _cfg()
+    inp = _inputs()
+    api_version = _api_version(inp, is_v2)
+    limit = inp.get("limit") or 50
+    start_time = inp.get("start_time")
+    end_time = inp.get("end_time")
+    secs = _rel_seconds(inp.get("time_range"))
+    if secs:
+        now = datetime.now(timezone.utc)
+        start_time = (now - timedelta(seconds=secs)).strftime(ISO)
+        end_time = now.strftime(ISO)
+    params = {
+        "index": inp.get("index") or "0",
+        "size": inp.get("page_size") or limit,
+        "statuses": inp.get("statuses"),
+        "start_time": start_time,
+        "end_time": end_time,
+    }
+    if api_version == "V2":
+        params.update({
+            "sources": inp.get("sources"),
+            "priorities": inp.get("priorities"),
+            "assignee_email": inp.get("assignee_email"),
+            "sort_field": inp.get("sort_field"),
+            "sort_direction": inp.get("sort_direction"),
+            "tags": inp.get("tags"),
+            "multi-customer": "true" if multi else None,
+        })
+    endpoint = "idr/" + api_version.lower() + "/investigations"
+    print(json.dumps(request("GET", endpoint, params=params)))
+
+
+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)

integrations/rapid7-insightidr/scripts/list_log_sets.py

@@ -0,0 +1,31 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    return base, headers
+
+
+def request(method, path):
+    base, headers = _cfg()
+    req = urllib.request.Request(base + path.lstrip("/"), headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    print(json.dumps(request("GET", "log_search/management/logsets")))
+
+
+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)

integrations/rapid7-insightidr/scripts/list_logs.py

@@ -0,0 +1,31 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    return base, headers
+
+
+def request(method, path):
+    base, headers = _cfg()
+    req = urllib.request.Request(base + path.lstrip("/"), headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    print(json.dumps(request("GET", "log_search/management/logs")))
+
+
+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)

integrations/rapid7-insightidr/scripts/list_users.py

@@ -0,0 +1,72 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+USER_SEARCH = ["first_name", "last_name", "name"]
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    return base, headers
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _list(v):
+    if v in (None, ""):
+        return []
+    if isinstance(v, list):
+        return [str(x).strip() for x in v if str(x).strip()]
+    return [p.strip() for p in str(v).split(",") if p.strip()]
+
+
+def request(method, path, params=None, body=None):
+    base, headers = _cfg()
+    url = base + path.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
+    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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    inp = _inputs()
+    rrn = inp.get("rrn")
+    if rrn:
+        print(json.dumps(request("GET", "idr/v1/users/" + urllib.parse.quote(rrn, safe=""))))
+        return
+    operator = inp.get("search_operator")
+    search = []
+    for field in USER_SEARCH:
+        values = _list(inp.get(field))
+        if values and not operator:
+            raise ValueError("search_operator is required to use first_name/last_name/name filters.")
+        for value in values:
+            search.append({"field": field, "operator": str(operator).upper(), "value": value})
+    direction = str(inp.get("sort_direction") or "asc").upper()
+    sort = [{"field": f, "order": direction} for f in _list(inp.get("sort"))]
+    params = {"index": inp.get("index") or "0", "size": inp.get("page_size") or inp.get("limit") or 50}
+    body = {}
+    if search:
+        body["search"] = search
+    if sort:
+        body["sort"] = sort
+    print(json.dumps(request("POST", "idr/v1/users/_search", params=params, body=body)))
+
+
+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)

integrations/rapid7-insightidr/scripts/query_log.py

@@ -0,0 +1,85 @@
+import json, os, sys, time, urllib.request, urllib.parse, urllib.error
+
+_UNITS = {"second": 1, "minute": 60, "hour": 3600, "day": 86400, "week": 604800, "month": 2592000, "year": 31536000}
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    return base, headers
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _rel_ms_range(text):
+    if not text:
+        return None, None
+    num = unit = None
+    for t in str(text).lower().replace("last", "").split():
+        if t.isdigit():
+            num = int(t)
+        elif t.rstrip("s") in _UNITS:
+            unit = t.rstrip("s")
+    if num is None or not unit:
+        return None, None
+    now_ms = int(time.time() * 1000)
+    return now_ms - num * _UNITS[unit] * 1000, now_ms
+
+
+def request(method, url, params=None):
+    base, headers = _cfg()
+    full = url if url.startswith("http") else base + url.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            full += ("&" if "?" in full else "?") + urllib.parse.urlencode(clean, doseq=True)
+    req = urllib.request.Request(full, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _collect(first):
+    # Follow the query's pagination links until no more pages remain.
+    events = []
+    queue = [first]
+    while queue:
+        page = queue.pop(0)
+        events.extend(page.get("events", []) or [])
+        for link in page.get("links", []) or []:
+            href = link.get("href")
+            if href:
+                queue.append(request("GET", href))
+    return events
+
+
+def run():
+    inp = _inputs()
+    start_time = inp.get("start_time")
+    end_time = inp.get("end_time")
+    if inp.get("time_range"):
+        start_time, end_time = _rel_ms_range(inp.get("time_range"))
+    params = {
+        "query": inp.get("query"),
+        "from": start_time,
+        "to": end_time,
+        "per_page": inp.get("logs_per_page"),
+        "sequence_number": inp.get("sequence_number"),
+    }
+    log_id = inp.get("log_id", "")
+    first = request("GET", "log_search/query/logs/" + urllib.parse.quote(log_id, safe=""), params=params)
+    print(json.dumps({"events": _collect(first)}))
+
+
+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)

integrations/rapid7-insightidr/scripts/query_log_set.py

@@ -0,0 +1,84 @@
+import json, os, sys, time, urllib.request, urllib.parse, urllib.error
+
+_UNITS = {"second": 1, "minute": 60, "hour": 3600, "day": 86400, "week": 604800, "month": 2592000, "year": 31536000}
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    return base, headers
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _rel_ms_range(text):
+    if not text:
+        return None, None
+    num = unit = None
+    for t in str(text).lower().replace("last", "").split():
+        if t.isdigit():
+            num = int(t)
+        elif t.rstrip("s") in _UNITS:
+            unit = t.rstrip("s")
+    if num is None or not unit:
+        return None, None
+    now_ms = int(time.time() * 1000)
+    return now_ms - num * _UNITS[unit] * 1000, now_ms
+
+
+def request(method, url, params=None):
+    base, headers = _cfg()
+    full = url if url.startswith("http") else base + url.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            full += ("&" if "?" in full else "?") + urllib.parse.urlencode(clean, doseq=True)
+    req = urllib.request.Request(full, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _collect(first):
+    events = []
+    queue = [first]
+    while queue:
+        page = queue.pop(0)
+        events.extend(page.get("events", []) or [])
+        for link in page.get("links", []) or []:
+            href = link.get("href")
+            if href:
+                queue.append(request("GET", href))
+    return events
+
+
+def run():
+    inp = _inputs()
+    start_time = inp.get("start_time")
+    end_time = inp.get("end_time")
+    if inp.get("time_range"):
+        start_time, end_time = _rel_ms_range(inp.get("time_range"))
+    params = {
+        "query": inp.get("query"),
+        "from": start_time,
+        "to": end_time,
+        "per_page": inp.get("logs_per_page"),
+        "sequence_number": inp.get("sequence_number"),
+    }
+    log_set_id = inp.get("log_set_id", "")
+    first = request("GET", "log_search/query/logsets/" + urllib.parse.quote(log_set_id, safe=""), params=params)
+    print(json.dumps({"events": _collect(first)}))
+
+
+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)

integrations/rapid7-insightidr/scripts/replace_threat_indicators.py

@@ -0,0 +1,73 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    return base, headers
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _list(v):
+    if v in (None, ""):
+        return []
+    if isinstance(v, list):
+        return [str(x).strip() for x in v if str(x).strip()]
+    return [p.strip() for p in str(v).split(",") if p.strip()]
+
+
+def _prune(d):
+    return {k: v for k, v in d.items() if v}
+
+
+def request(method, path, params=None, body=None):
+    base, headers = _cfg()
+    url = base + path.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
+    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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+# replace deletes the threat's existing indicators before adding the supplied ones.
+ACTION = "replace"
+
+
+def run():
+    inp = _inputs()
+    body = _prune({
+        "ips": _list(inp.get("ip_addresses")),
+        "hashes": _list(inp.get("hashes")),
+        "domain_names": _list(inp.get("domain_names")),
+        "urls": _list(inp.get("url")),
+    })
+    results = []
+    for key in _list(inp.get("key")):
+        out = request(
+            "POST",
+            "idr/v1/customthreats/key/" + urllib.parse.quote(key, safe="") + "/indicators/" + ACTION,
+            params={"format": "json"},
+            body=body,
+        )
+        results.append(out.get("threat", out))
+    print(json.dumps({"data": results}))
+
+
+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)

integrations/rapid7-insightidr/scripts/search_investigation.py

@@ -0,0 +1,92 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+CONTAINS = "CONTAINS"
+EQUALS = "EQUALS"
+# (field, operator) pairs mirrored from the InsightIDR investigation search schema.
+SEARCH = [
+    ("actor_asset_hostname", CONTAINS),
+    ("actor_user_name", CONTAINS),
+    ("alert_mitre_t_codes", EQUALS),
+    ("alert_rule_rrn", EQUALS),
+    ("assignee_id", EQUALS),
+    ("organization_id", EQUALS),
+    ("priority", EQUALS),
+    ("rrn", EQUALS),
+    ("source", EQUALS),
+    ("status", EQUALS),
+    ("title", CONTAINS),
+]
+SORT_FIELDS = {
+    "Created time": "created_time",
+    "Priority": "priority",
+    "RRN": "rrn",
+    "Alert created time": "alerts_most_recent_created_time",
+    "Alert detection created time": "alerts_most_recent_detection_created_time",
+}
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    multi = str(s.get("is_multi_customer") or "").strip().lower() in ("1", "true", "yes")
+    return base, headers, multi
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _list(v):
+    if v in (None, ""):
+        return []
+    if isinstance(v, list):
+        return [str(x).strip() for x in v if str(x).strip()]
+    return [p.strip() for p in str(v).split(",") if p.strip()]
+
+
+def request(method, path, params=None, body=None):
+    base, headers, _ = _cfg()
+    url = base + path.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
+    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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    _, _, multi = _cfg()
+    inp = _inputs()
+    search = []
+    for field, op in SEARCH:
+        for value in _list(inp.get(field)):
+            search.append({"field": field, "operator": op, "value": value})
+    direction = str(inp.get("sort_direction") or "asc").upper()
+    sort = [{"field": SORT_FIELDS.get(f, f), "order": direction} for f in _list(inp.get("sort"))]
+    body = {"search": search, "sort": sort}
+    if inp.get("start_time"):
+        body["start_time"] = inp["start_time"]
+    if inp.get("end_time"):
+        body["end_time"] = inp["end_time"]
+    params = {
+        "index": inp.get("index") or "0",
+        "size": inp.get("page_size") or inp.get("limit") or 50,
+        "multi-customer": "true" if multi else None,
+    }
+    print(json.dumps(request("POST", "idr/v2/investigations/_search", params=params, body=body)))
+
+
+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)

integrations/rapid7-insightidr/scripts/set_status.py

@@ -0,0 +1,85 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    multi = str(s.get("is_multi_customer") or "").strip().lower() in ("1", "true", "yes")
+    is_v2 = str(s.get("is_v2") or "true").strip().lower() in ("1", "true", "yes")
+    return base, headers, multi, is_v2
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _api_version(inp, is_v2):
+    v = str(inp.get("api_version") or "Default").strip()
+    return v if v in ("V1", "V2") else ("V2" if is_v2 else "V1")
+
+
+def _list(v):
+    if v in (None, ""):
+        return []
+    if isinstance(v, list):
+        return [str(x).strip() for x in v if str(x).strip()]
+    return [p.strip() for p in str(v).split(",") if p.strip()]
+
+
+def _prune(d):
+    return {k: v for k, v in d.items() if v not in (None, "", {}, [])}
+
+
+def _camel(text):
+    return "".join(w.capitalize() for w in str(text).split()) if text else None
+
+
+def request(method, path, params=None, body=None):
+    base, headers, _, _ = _cfg()
+    url = base + path.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
+    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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    _, _, multi, is_v2 = _cfg()
+    inp = _inputs()
+    api_version = _api_version(inp, is_v2)
+    status = inp.get("status", "")
+    body = {}
+    params = {}
+    if api_version == "V2":
+        body = _prune({
+            "disposition": inp.get("disposition"),
+            "threat_command_close_reason": _camel(inp.get("threat_command_close_reason")),
+            "threat_command_free_text": inp.get("threat_command_free_text"),
+        })
+        params = {"multi-customer": "true" if multi else None}
+    results = []
+    for inv_id in _list(inp.get("investigation_id")):
+        results.append(request(
+            "PUT",
+            "idr/" + api_version.lower() + "/investigations/" + urllib.parse.quote(inv_id, safe="") + "/status/" + urllib.parse.quote(status, safe=""),
+            params=params,
+            body=body or None,
+        ))
+    print(json.dumps({"data": results}))
+
+
+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)

integrations/rapid7-insightidr/scripts/test_connection.py

@@ -0,0 +1,46 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    multi = str(s.get("is_multi_customer") or "").strip().lower() in ("1", "true", "yes")
+    is_v2 = str(s.get("is_v2") or "true").strip().lower() in ("1", "true", "yes")
+    return base, headers, multi, is_v2
+
+
+def request(method, path, params=None, body=None):
+    base, headers, _, _ = _cfg()
+    url = base + path.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
+    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 r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    request("GET", "idr/v1/investigations", params={"size": 1})
+    print(json.dumps({"ok": True}))
+
+
+try:
+    run()
+except urllib.error.HTTPError as e:
+    detail = e.read().decode("utf-8", "replace")
+    if e.code == 401:
+        print(json.dumps({"ok": False, "error": "API key is not valid."}))
+    elif e.code == 500:
+        print(json.dumps({"ok": False, "error": "Wrong account region."}))
+    else:
+        print(json.dumps({"ok": False, "error": "HTTP " + str(e.code), "detail": detail}))
+    sys.exit(1)
+except Exception as e:
+    print(json.dumps({"ok": False, "error": str(e)}))
+    sys.exit(1)

integrations/rapid7-insightidr/scripts/update_investigation.py

@@ -0,0 +1,65 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+
+def _cfg():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    region = str(s.get("region") or "us").strip().lower()
+    base = "https://" + region + ".api.insight.rapid7.com/"
+    headers = {"X-Api-Key": s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"}
+    multi = str(s.get("is_multi_customer") or "").strip().lower() in ("1", "true", "yes")
+    return base, headers, multi
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def _prune(d):
+    return {k: v for k, v in d.items() if v not in (None, "", {}, [])}
+
+
+def _camel(text):
+    return "".join(w.capitalize() for w in str(text).split()) if text else None
+
+
+def request(method, path, params=None, body=None):
+    base, headers, _ = _cfg()
+    url = base + path.lstrip("/")
+    if params:
+        clean = {k: v for k, v in params.items() if v not in (None, "")}
+        if clean:
+            url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
+    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=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def run():
+    _, _, multi = _cfg()
+    inp = _inputs()
+    inv_id = inp.get("investigation_id", "")
+    disposition = inp.get("disposition")
+    body = _prune({
+        "assignee": _prune({"email": inp.get("user_email_address")}),
+        "disposition": disposition.replace(" ", "_") if disposition else None,
+        "priority": inp.get("priority"),
+        "status": inp.get("status"),
+        "title": inp.get("title"),
+        "threat_command_close_reason": _camel(inp.get("threat_command_close_reason")),
+        "threat_command_free_text": inp.get("threat_command_free_text"),
+    })
+    out = request("PATCH", "idr/v2/investigations/" + urllib.parse.quote(inv_id, safe=""),
+                  params={"multi-customer": "true" if multi else None}, body=body)
+    print(json.dumps(out))
+
+
+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)