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 _rrn_region(rrn):
    """The region segment of an RRN, or None.

    An RRN is rrn:<service>:<region>:<org>:<type>:… — so a resource states which
    region it lives in, while the base URL is built from the instance's
    configured region. Those two agree on a single-region tenant and need not on
    a multi-customer key, which spans organizations.
    """
    parts = str(rrn or "").split(":")
    return parts[2] if len(parts) > 4 and parts[0] == "rrn" and parts[2] else None


def _collect_region_failures(out, sink):
    """Carry a partial failure out of a response instead of dropping it.

    The Alert Triage API answers HTTP 200 with a region_failures list when it
    could not reach one of the regions behind the request — the spec says the
    presence of items there IS the partial failure. Reading only the payload
    turns that into missing alerts with no error anywhere, which for an
    ingestion feed is indistinguishable from a quiet period.
    """
    for f in (out.get("region_failures") or []):
        sink.append(f)
    return out


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 = [], [], {}
    region_failures = []
    index = int(inp.get("index") or 0)
    while len(alerts) + len(rrns) < limit:
        size = min(PAGE, limit - len(alerts) - len(rrns))
        out = _collect_region_failures(
            request("POST", AT + "alerts/ops/search",
                    params={"index": index, "size": size,
                            "rrns_only": "true" if rrns_only else None},
                    body=body),
            region_failures)
        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

    if not isinstance(meta, dict):
        meta = {}
    # A non-empty region_failures means alerts from that region are missing from
    # this page, on an otherwise successful response.
    meta["region_failures"] = region_failures
    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)