riposte-marketplace/integrations/rapid7-insightidr/scripts/search_alerts.py

import json, os, sys, urllib.request, urllib.parse, urllib.error

# The Alert Triage API lives under /idr/at, a base of its own — not /idr/v1 or
# /idr/v2 like the investigation endpoints. Same host, same API key.
AT = "idr/at/"


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 {}

from datetime import datetime, timedelta, timezone

ISO = "%Y-%m-%dT%H:%M:%SZ"
PAGE = 100


def _terms(inp):
    """Build the search terms array.

    Two ways in, because the field ids are per-tenant and discovered at runtime
    (see list_alert_fields): a raw JSON passthrough for full control, and one
    field/operator/values triple for the common case.
    """
    raw = inp.get("terms_json")
    if raw:
        if isinstance(raw, str):
            try:
                parsed = json.loads(raw)
            except ValueError as e:
                # json's own message ("Expecting value: line 1 column 1") names
                # neither the input nor what it received, so an operator who
                # pasted the field name along with the value — the obvious
                # mistake, since both are shown together — gets a parser error
                # with nothing to act on. Echo the head of what arrived.
                head = raw[:80] + ("…" if len(raw) > 80 else "")
                raise ValueError("terms_json is not valid JSON (" + str(e) + "). "
                                 "Expected a JSON array of search terms, got: " + head)
            return parsed if isinstance(parsed, list) else [parsed]
        return raw if isinstance(raw, list) else [raw]
    field_ids = _list(inp.get("search_field"))
    values = _list(inp.get("search_values"))
    if not field_ids or not values:
        # The schema marks terms as required. An empty array is what "no filter"
        # has to look like; if the tenant refuses it, the HTTP error says so
        # rather than this script inventing a filter nobody asked for.
        return []
    op = str(inp.get("search_operator") or "EQUALS").strip().upper()
    return [{"field_ids": field_ids, "operator": op, "terms": values}]


def run():
    inp = _inputs()
    limit = int(inp.get("limit") or 50)
    start_time = inp.get("start_time")
    end_time = inp.get("end_time")
    if not start_time:
        # start_time is required by the API. Default to a window rather than
        # failing: an ingestion rule that has never run has no watermark yet.
        days = int(inp.get("default_days") or 1)
        start_time = (datetime.now(timezone.utc) - timedelta(days=days)).strftime(ISO)
    search = {"start_time": start_time, "terms": _terms(inp)}
    if end_time:
        search["end_time"] = end_time
    if inp.get("leql"):
        search["leql"] = inp["leql"]

    body = {"search": search}
    ids = _list(inp.get("field_ids"))
    if ids:
        # Without field_ids the alert carries no additional fields at all — this
        # is where the hostnames, accounts and command lines come from.
        body["field_ids"] = ids
    sort_field = inp.get("sort_field")
    if sort_field:
        body["sorts"] = [{"field_id": sort_field,
                          "order": str(inp.get("sort_direction") or "ASC").strip().upper()}]

    rrns_only = str(inp.get("rrns_only") or "").strip().lower() in ("1", "true", "yes")
    alerts, rrns, meta = [], [], {}
    index = int(inp.get("index") or 0)
    while len(alerts) + len(rrns) < limit:
        size = min(PAGE, limit - len(alerts) - len(rrns))
        out = request("POST", AT + "alerts/ops/search",
                      params={"index": index, "size": size,
                              "rrns_only": "true" if rrns_only else None},
                      body=body)
        meta = out.get("metadata") or {}
        page_alerts = out.get("alerts") or []
        page_rrns = out.get("rrns") or []
        alerts.extend(page_alerts)
        rrns.extend(page_rrns)
        if meta.get("is_last_index") or not (page_alerts or page_rrns):
            break
        index += 1

    result = {"metadata": meta}
    if rrns_only:
        result["rrns"] = rrns[:limit]
    else:
        result["alerts"] = alerts[:limit]
    print(json.dumps(result))


try:
    run()
except urllib.error.HTTPError as e:
    detail = e.read().decode("utf-8", "replace")
    if e.code in (401, 403):
        detail += (" — the Alert Triage API (/idr/at) is only available to Managed Detection and"
                   " Response customers with access to the Alerts experience; a key that works on"
                   " the investigation endpoints can still be refused here")
    print(json.dumps({"error": "HTTP " + str(e.code), "detail": detail}))
    sys.exit(1)
except Exception as e:
    print(json.dumps({"error": str(e)}))
    sys.exit(1)