rapid7-insightidr 1.14.7: user_authentications, one call for a UBA alert

A new read command answers what a behaviour authentication alert asks —
which assets the user authenticated to around the alert, and which were new —
in one call: it resolves the Asset Authentication log set, builds the
keyword query and the 24 h window itself, and summarizes all events and,
apart, those marked new_authentication. Across real runs gpt-oss mangled log
set ids, computed the window a day off, put a filter inside the LEQL (HTTP
400) and read ten lines out of a thousand; this leaves it nothing to get
wrong. Static calls it first for such alerts; list_investigation_alerts
points to it.

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

integrations/rapid7-insightidr/agents/operator.yaml

@@ -36,8 +36,15 @@ instructions: |
      investigation RRN (rrn:investigation:...), never an alert RRN.
   2. list_investigation_alerts(investigation_id=<that RRN>) (1 call).
   3. Alerts with evidence_api true: get_alert_actors, then get_alert_evidence.
-  4. Alerts with evidence_api false, and any alert whose evidence did not
+  4. A behaviour (UBA) alert about authentication — lateral movement, "new
-     answer: the logs.
+     assets", "Domain Credentials": call user_authentications(user=<the
+     account>, anchor_time=<the alert's first seen time, copied as given>)
+     ONCE. It does the search, the window and the split itself. Report its
+     "window" word for word, the totals from "all", and from
+     "new_authentications" the destination assets by name with their counts:
+     those are the new assets. Do not redo this search with query_log_set.
+     Other alerts with evidence_api false, or a question this does not
+     answer: the logs, as below.
      - list_log_sets (1 call) to see the names. Then query_log_set with
        log_set_name — the NAME, e.g. "Asset Authentication" — never a copied
        log_set_id: a 36-character id gets mangled in the copy.
@@ -80,6 +87,7 @@ instructions: |
   assets" the alert is about.
 
 commands:
+  - user_authentications
   - list_investigation_alerts
   - get_investigation
   - search_investigation

integrations/rapid7-insightidr/scripts/list_investigation_alerts.py

@@ -65,9 +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: query_log_set with log_set_name \"Asset Authentication\" and "
+                     "authentication logs: call user_authentications(user=<the account>, anchor_time=<this alert's "
-                     "where(\"<user>\") and anchor_time = this alert's first seen time: the 24 hours before it "
+                     "first seen time, as given>) once — it searches, sets the window and separates the new "
-                     "are queried — a behaviour alert aggregates hours of activity.")
+                     "assets itself.")
 
 
 try:

integrations/rapid7-insightidr/scripts/user_authentications.py

@@ -0,0 +1,312 @@
+import datetime, hashlib, 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}
+
+# Beyond this a single value can only be milliseconds: 1e11 seconds is the year 5138,
+# and 1e11 milliseconds was 1973. Nothing a feed can legitimately ask for sits between.
+_MS_FLOOR = 100000000000
+
+
+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 _to_ms(v):
+    """Epoch milliseconds, whatever the caller wrote.
+
+    As an ingestion feed this command receives start_time from the scheduler's
+    high-water mark, and the mark's format is an operator setting on the rule
+    (iso8601, unix, unix_ms). The API only takes milliseconds. Reading all three
+    here is what keeps the feed from silently fetching the wrong window — or
+    nothing at all — because of a dropdown three screens away.
+    """
+    if v in (None, ""):
+        return None
+    s = str(v).strip()
+    if s.isdigit():
+        n = int(s)
+        return n if n >= _MS_FLOOR else n * 1000
+    try:
+        return int(datetime.datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
+    except Exception:
+        return None
+
+
+def _num(v, default):
+    try:
+        return float(v) if v not in (None, "") else default
+    except (TypeError, ValueError):
+        return default
+
+
+def _iso(ms):
+    """The window as it was queried, in the form the agent reports it."""
+    return datetime.datetime.fromtimestamp(ms / 1000, tz=datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+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, limit):
+    """Walk the pagination links, stopping at `limit`.
+
+    The walk is bounded because this runs unattended every few minutes: a query
+    that matches more than it was meant to would otherwise hold the run open for
+    as many pages as the window contains, and the next tick would find the rule
+    still in flight and skip itself.
+    """
+    events = []
+    queue = [first]
+    while queue and len(events) < limit:
+        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[:limit]
+
+
+def _identify(events, fallback_log_id, detection_name):
+    """Give every event a stable id, and the name of the rule it stands for.
+
+    dedup_key needs a field that is identical across runs for the same log line.
+    The scheduler deliberately re-reads the last minute of the previous window on
+    every run (so an event written while a poll was in flight is not lost), which
+    means the same line WILL come back — anything derived from the fetch time, or
+    from the position in the page, would make it a second incident every time.
+    log_id + sequence_number is the line's own address in Log Search; the hash is
+    the fallback for a source that does not number its entries.
+
+    detection_name is stamped here because the log line has no idea which basic
+    detection rule it is being replayed for. Without it an incident is titled by
+    its own first characters, and an analyst reads a log line where the source
+    console shows an alert name.
+    """
+    out = []
+    for e in events:
+        if not isinstance(e, dict):
+            continue
+        log_id = e.get("log_id") or fallback_log_id or ""
+        seq = e.get("sequence_number")
+        if seq in (None, ""):
+            digest = hashlib.sha1(
+                ("%s|%s|%s" % (log_id, e.get("timestamp"), e.get("message"))).encode("utf-8", "replace")
+            ).hexdigest()
+            e["event_id"] = "%s:%s" % (log_id, digest)
+        else:
+            e["event_id"] = "%s:%s" % (log_id, seq)
+        if not e.get("log_id"):
+            e["log_id"] = log_id
+        if detection_name:
+            e["detection_name"] = detection_name
+        out.append(e)
+    return out
+
+
+# What a summary keeps: the values an analyst reads first, per field.
+_TOP_VALUES = 8
+_MAX_FIELDS = 30
+_SAMPLES = 5
+# Rapid7's whole vendor payload, kilobytes per event: it stays out of a summary.
+_SKIP_FIELDS = {"source_json", "source_data"}
+
+
+def _fields(e):
+    """One Log Search event as flat fields. `message` is a JSON-encoded string
+    on parsed sources; nested objects are flattened one level (a.b)."""
+    msg = e.get("message")
+    data = None
+    if isinstance(msg, str):
+        try:
+            data = json.loads(msg)
+        except ValueError:
+            data = None
+    elif isinstance(msg, dict):
+        data = msg
+    if not isinstance(data, dict):
+        return {"message": msg} if msg not in (None, "") else {}
+    flat = {}
+    for k, v in data.items():
+        if k in _SKIP_FIELDS:
+            continue
+        if isinstance(v, dict):
+            for k2, v2 in v.items():
+                if not isinstance(v2, (dict, list)) and v2 not in (None, ""):
+                    flat[k + "." + k2] = v2
+        elif not isinstance(v, list) and v not in (None, ""):
+            flat[k] = v
+    return flat
+
+
+def _parse_only(v):
+    """"field=value, field2=value2" -> {field: value}. Applied to the events'
+    fields after the search, so narrowing a result needs no query syntax."""
+    out = {}
+    for part in str(v or "").split(","):
+        if "=" in part:
+            k, val = part.split("=", 1)
+            if k.strip():
+                out[k.strip()] = val.strip().strip("\"'")
+    return out
+
+
+def _matches(fields, only):
+    """Every condition holds, comparing as text and ignoring case."""
+    for k, want in only.items():
+        if str(fields.get(k, "")).lower() != want.lower():
+            return False
+    return True
+
+
+def _summarize(events, limit):
+    """Counts instead of lines: per field, its most frequent values.
+
+    An agent asked who a user logged on to does not need a thousand raw lines —
+    on a real run it got 1000 events, could read about ten, and reported the
+    accounts and hosts of those ten. Top values per field answer the question
+    over the whole window and fit in a model's context.
+    """
+    counts = {}
+    first_ts, last_ts = None, None
+    for e in events:
+        ts = e.get("timestamp")
+        if isinstance(ts, (int, float)):
+            first_ts = ts if first_ts is None or ts < first_ts else first_ts
+            last_ts = ts if last_ts is None or ts > last_ts else last_ts
+        for k, v in _fields(e).items():
+            counts.setdefault(k, {})
+            key = str(v)[:200]
+            counts[k][key] = counts[k].get(key, 0) + 1
+    # Fields that vary least first: they are the ones that describe the activity
+    # (result, service, destination) rather than identify a line (timestamps, ids).
+    ranked = sorted(counts.items(), key=lambda kv: (len(kv[1]), kv[0]))
+    top = {}
+    for k, vals in ranked[:_MAX_FIELDS]:
+        top[k] = {"distinct": len(vals),
+                  "top": [[v, n] for v, n in sorted(vals.items(), key=lambda x: -x[1])[:_TOP_VALUES]]}
+    return {
+        "count": len(events),
+        "truncated": len(events) >= limit,
+        "first_event": _iso(first_ts) if first_ts is not None else None,
+        "last_event": _iso(last_ts) if last_ts is not None else None,
+        "fields": top,
+        "fields_omitted": max(0, len(counts) - _MAX_FIELDS),
+        "samples": [_fields(e) for e in events[:_SAMPLES]],
+    }
+
+
+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)))
+
+
+NEW_MARKERS = ("new_authentication",)
+
+
+def run():
+    """One call for the question a behaviour (UBA) authentication alert asks:
+    which assets did this user authenticate to around the alert, and which of
+    them were new. The agent gives a user and the alert's time; the log set,
+    the query, the window and the split are done here — on real runs the model
+    mangled log set ids, computed the window a day off, put a filter inside the
+    LEQL and got a 400, and read ten lines out of a thousand."""
+    inp = _inputs()
+    user = str(inp.get("user") or "").strip().strip("\"'")
+    if not user or '"' in user:
+        raise ValueError("user is required: the account name as the alert gives it, e.g. jdoe")
+    anchor = _to_ms(inp.get("anchor_time"))
+    if anchor is None:
+        raise ValueError("anchor_time is required: the alert's first seen time, exactly as the data gives it")
+    before = _num(inp.get("hours_before"), 24.0)
+    after = _num(inp.get("minutes_after"), 30.0)
+    start_time = int(anchor - before * 3600 * 1000)
+    end_time = int(anchor + after * 60 * 1000)
+    name = str(inp.get("log_set_name") or "Asset Authentication").strip()
+    log_set_id = _resolve_name(name)
+    try:
+        limit = int(inp.get("limit") or 2000)
+    except (TypeError, ValueError):
+        limit = 2000
+    query = 'where("%s")' % user
+    first = request("GET", "log_search/query/logsets/" + urllib.parse.quote(log_set_id, safe=""),
+                    params={"query": query, "from": start_time, "to": end_time})
+    events = _identify(_collect(first, limit), "", None)
+    new = [e for e in events
+           if any(str(v).lower() in NEW_MARKERS for v in _fields(e).values())]
+    print(json.dumps({
+        "user": user,
+        "log_set": name,
+        "query": query,
+        "window": "%s to %s" % (_iso(start_time), _iso(end_time)),
+        "all": _summarize(events, limit),
+        "new_authentications": _summarize(new, limit),
+        "note": ("new_authentications holds only the events Rapid7 marked new_authentication: their "
+                 "destination assets are the new assets a behaviour alert counts."),
+    }))
+
+
+try:
+    run()
+except urllib.error.HTTPError as e:
+    out = {"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}
+    if e.code == 404:
+        out["hint"] = "No log set with this log_set_id in the configured region. Check the ID against list_log_sets, and the instance region."
+    print(json.dumps(out))
+    sys.exit(1)
+except Exception as e:
+    print(json.dumps({"error": str(e)}))
+    sys.exit(1)