feat(rapid7-insightidr): fetch and parse the evidence behind each alert (v1.8.0)

An alert names what fired. The evidence is what happened, and until now it was
never fetched — which is why an incident could show a rule name and nothing
else. On a Fortigate admin login the alert carried an empty
rule_keys_of_interest; the evidence carried the whole event.

It cannot be reached by mapping alone. The evidence's `data` is a
JSON-ENCODED STRING holding {source_data, custom_data}, and JSONata does not
deserialise, so no mapper expression can ever see inside it. The parsing has to
happen in the command, and that is the reason for this change more than the
extra call is.

Once parsed, two views of the same event:

  custom_data  the vendor's normalised fields, prefixed per source
               (Forti_user, Forti_srcip) — 6 keys on the sample
  source_data  the raw line in key="value" form — 25 keys

Both are merged into evidence_fields with the vendor prefix stripped, so a
mapper names `user` once instead of once per vendor, and source_data is overlaid
last because it carries more and is not affected by the quoting artefact below.

Values are cleaned, and that is not cosmetic: custom_data returns
'10.70.20.66"' with a trailing quote left by the vendor parser. Kept as-is it
yields an address matching no regex — an observable nobody can pivot on, and no
error to say why.

The line itself lands on raw_data and the source's own sentence on message,
both of which the layout builder offers. The Fortigate login now maps to
src_endpoint.ip, dst_endpoint.ip, actor.user.name, device.hostname/name/uid,
activity_name "login" and status_detail "success", and indexes four entities
where it indexed none.

Optional (include_evidence, default true) and capped (max_evidence_per_alert,
default 5), one call per alert under the run's existing time budget. An alert
whose evidence cannot be read keeps an evidence_error and still arrives;
evidence_calls and evidence_errors are reported in the run metadata.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

integrations/rapid7-insightidr/scripts/get_investigations_full.py

@@ -43,6 +43,7 @@ def request(method, path, params=None, body=None):
         raw = r.read()
         return json.loads(raw) if raw else {}
 
+import re
 import time
 from datetime import datetime, timedelta, timezone
 
@@ -112,6 +113,62 @@ def _collect_region_failures(out, sink):
     return out
 
 
+# key="value" / key=value, the shape syslog-derived log lines take. Non-matching
+# sources simply yield nothing, which is the right answer for them.
+_KV = re.compile(r'([A-Za-z_][\w.\-]*)=("(?:[^"\\]|\\.)*"|\S+)')
+
+
+def _clean(v):
+    """Strip the quoting Rapid7's parser leaves behind.
+
+    custom_data values arrive as '10.70.20.66"' — a trailing quote from the
+    vendor parser. Left alone it produces an IP that matches no regex and an
+    entity nobody can pivot on, so the cleanup is not cosmetic.
+    """
+    v = str(v).strip()
+    if len(v) >= 2 and v[0] == '"' and v[-1] == '"':
+        v = v[1:-1]
+    return v.replace('\\"', '"').strip().strip('"')
+
+
+def _evidence_fields(ev):
+    """One evidence's log line as a flat, normalised dict.
+
+    `data` is a JSON-encoded STRING holding {source_data, custom_data} — the
+    actual event behind the alert. JSONata cannot deserialise, so a mapper can
+    never reach into it; this is why the parsing has to happen here.
+
+    custom_data is the vendor's normalised view, its keys prefixed per source
+    (Forti_user, Forti_srcip). The prefix is stripped so a mapper can name
+    `user` once rather than once per vendor. source_data is the raw line and is
+    overlaid on top: it carries far more keys (25 against 6 on the sample this
+    was written from) and it is not affected by the quoting artefact above.
+    """
+    raw = ev.get("data")
+    if not isinstance(raw, str) or not raw.strip():
+        return {}, None
+    try:
+        blob = json.loads(raw)
+    except ValueError:
+        # Not JSON: keep the string itself as the raw line rather than lose it.
+        return {}, raw
+    if not isinstance(blob, dict):
+        return {}, raw
+    fields = {}
+    custom = blob.get("custom_data")
+    if isinstance(custom, dict):
+        for k, v in custom.items():
+            if not isinstance(v, (str, int, float)):
+                continue
+            key = k.split("_", 1)[1] if "_" in k else k
+            fields[key.lower()] = _clean(v)
+    source = blob.get("source_data")
+    if isinstance(source, str):
+        for k, v in _KV.findall(source):
+            fields[k.lower()] = _clean(v)
+    return fields, (source if isinstance(source, str) else raw)
+
+
 def run():
     _, _, multi = _cfg()
     inp = _inputs()
@@ -185,6 +242,34 @@ def run():
             if a.get("rrn"):
                 by_rrn[a["rrn"]] = a
 
+    # ④ The evidence behind each alert: the actual log line. An alert names what
+    #    fired; the evidence is what happened. One call per alert — there is no
+    #    bulk form — so it is optional, capped, and under the same time budget.
+    evidence_calls, evidence_errors = 0, 0
+    if str(inp.get("include_evidence") or "true").strip().lower() in ("1", "true", "yes"):
+        per_alert = int(inp.get("max_evidence_per_alert") or 5)
+        for rrn, a in by_rrn.items():
+            if time.monotonic() - started > budget:
+                truncated = True
+                break
+            try:
+                page = request("GET", AT + "alerts/" + urllib.parse.quote(rrn, safe="") + "/evidences",
+                               params={"index": "0", "size": per_alert})
+                evidence_calls += 1
+            except urllib.error.HTTPError as e:
+                a["evidence_error"] = "HTTP " + str(e.code)
+                evidence_errors += 1
+                continue
+            evs = (page.get("evidences") or [])[:per_alert]
+            for ev in evs:
+                ev["fields"], ev["raw_line"] = _evidence_fields(ev)
+            a["evidences"] = evs
+            if evs:
+                # Lifted for the mapper: MapFields cannot index an array, and the
+                # ocsf section reads alerts[0] only.
+                a["evidence_fields"] = evs[0].get("fields") or {}
+                a["evidence_raw_line"] = evs[0].get("raw_line")
+
     fetched = 0
     for inv in investigations:
         rrns = rrns_by_inv.get(inv.get("rrn") or "", [])
@@ -218,6 +303,10 @@ def run():
                 "first_alert_logset_id": log.get("logset_id"),
                 "first_alert_log_entry_id": log.get("log_entry_id"),
             })
+            ef = a.get("evidence_fields") or {}
+            if ef:
+                inv["first_evidence_message"] = ef.get("msg") or ef.get("logdesc") or ef.get("message")
+                inv["first_evidence_raw_line"] = a.get("evidence_raw_line")
 
     print(json.dumps({
         "data": investigations,
@@ -225,6 +314,8 @@ def run():
             "investigations_returned": len(investigations),
             "alert_rrns_collected": len(order),
             "alerts_fetched": fetched,
+            "evidence_calls": evidence_calls,
+            "evidence_errors": evidence_errors,
             "alert_ids_without_rrn": skipped,
             "truncated": truncated,
             # HTTP 200 with a non-empty region_failures is a PARTIAL result: