rapid7-insightidr 1.14.2: query a log set by name; Static widens UBA windows

query_log_set accepts log_set_name and resolves the id itself (exact name, or
the one set whose name contains it; otherwise refused with the existing
names): on a real run the model mangled a 36-character id it copied. Static
queries by name, retries a 0-event keyword search case-insensitively, takes
the 24 hours before a behaviour alert (they aggregate hours of activity), and
always looks for the user's other investigations.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015RhL2gy6Sx6Y3FEsyBE3mZ

integrations/rapid7-insightidr/scripts/list_investigation_alerts.py

@@ -65,8 +65,9 @@ def _annotate(a):
     else:
         a["next"] = ("Behaviour (UBA) alert with no RRN: get_alert, get_alert_actors, get_alert_evidence and "
                      "get_alert_process_trees answer 404 for it — do not call them. Its evidence is in the "
-                     "authentication logs: list_log_sets, then query_log_set on the Asset Authentication log set "
-                     "(or each Active Directory one) with where(\"<user>\") over the alert's window.")
+                     "authentication logs: query_log_set with log_set_name \"Asset Authentication\" and "
+                     "where(\"<user>\"), over the 24 hours before the alert's first seen time — a behaviour "
+                     "alert aggregates hours of activity.")
 
 
 try:

integrations/rapid7-insightidr/scripts/query_log_set.py

@@ -125,6 +125,31 @@ def _identify(events, fallback_log_id, detection_name):
     return out
 
 
+def _log_sets():
+    """Every log set of the account, as (id, name)."""
+    out = request("GET", "log_search/management/logsets")
+    sets = out.get("logsets", []) if isinstance(out, dict) else []
+    return [(x.get("id", ""), x.get("name", "")) for x in sets if isinstance(x, dict)]
+
+
+def _resolve_name(name):
+    """The id of the log set called name: exact match first, ignoring case, then
+    the one log set whose name contains it. Anything else is refused with the
+    names that do exist, so the next call can pick one rather than guess."""
+    sets = _log_sets()
+    want = name.lower()
+    exact = [i for i, n in sets if n.lower() == want]
+    if len(exact) == 1:
+        return exact[0]
+    partial = [(i, n) for i, n in sets if want in n.lower()]
+    if len(partial) == 1:
+        return partial[0][0]
+    names = sorted(n for _, n in sets)
+    if partial:
+        raise ValueError("log_set_name %r matches several log sets: %s" % (name, ", ".join(sorted(n for _, n in partial))))
+    raise ValueError("no log set is called %r. Log sets: %s" % (name, ", ".join(names)))
+
+
 def run():
     inp = _inputs()
     start_time = _to_ms(inp.get("start_time"))
@@ -152,8 +177,14 @@ def run():
     # A pasted ID often drags a quote or a space along; quoted into the path it
     # names no log set, and the API answers a bare 404.
     log_set_id = str(inp.get("log_set_id") or "").strip().strip("\"'").strip()
+    name = str(inp.get("log_set_name") or "").strip()
+    if name:
+        # A model copying a 36-character id from one answer into the next call
+        # drops or swaps characters — seen on a real run, where the "Asset
+        # Authentication" query went to a mangled id. A name it copies right.
+        log_set_id = _resolve_name(name)
     if not log_set_id:
-        raise ValueError("log_set_id is required (list_log_sets gives it)")
+        raise ValueError("log_set_id or log_set_name is required (list_log_sets gives both)")
     first = request("GET", "log_search/query/logsets/" + urllib.parse.quote(log_set_id, safe=""), params=params)
     # No fallback log id: a log set spans several logs and every event names its own.
     events = _identify(_collect(first, limit), "", inp.get("detection_name"))