feat(sekoia-xdr): list_alerts attaches each alert's events (1.3.0)

An alert only summarises its detection; addresses, user, process and
command line live in the events tagged with its short id. with_events
runs one event search per alert (alert_short_ids, first_seen_at - 1 min
to now, as Sekoia's XSOAR integration does), in parallel within the
60 s command budget, and nests the dotted ECS names. A search not done
in time leaves events_pending + events_job_uuid: the alert still goes
out, as the rule's watermark would not offer it again.

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

integrations/sekoia-xdr/scripts/list_alerts.py

@@ -1,4 +1,11 @@
-import json, os, sys, urllib.request, urllib.parse, urllib.error
+import json, os, sys, time, urllib.request, urllib.parse, urllib.error
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timedelta, timezone
+
+# Riposte stops a command after 60 s. Event searches get what is left of this
+# budget once the alerts are listed; the margin covers container start-up.
+STARTED = time.monotonic()
+BUDGET = 45
 
 
 def _cfg():
@@ -12,19 +19,94 @@ def _inputs():
     return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
 
 
-def request(method, path, params=None):
+def left():
+    return BUDGET - (time.monotonic() - STARTED)
+
+
+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 += "?" + urllib.parse.urlencode(clean, doseq=True)
-    req = urllib.request.Request(url, headers=headers, method=method)
-    with urllib.request.urlopen(req, timeout=90) as r:
+    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=max(5, min(30, left()))) as r:
         raw = r.read()
         return json.loads(raw) if raw else {}
 
 
+def nest(event):
+    """Sekoia returns events flat, ECS names as dotted keys ("source.ip").
+    Rebuild the tree so a mapper reads events[0].source.ip; a key whose prefix
+    is already a plain value stays flat rather than overwrite it."""
+    out = {}
+    for key in sorted(event, key=lambda k: k.count(".")):
+        node, parts = out, key.split(".")
+        for p in parts[:-1]:
+            if p not in node:
+                node[p] = {}
+            if not isinstance(node[p], dict):
+                node = None
+                break
+            node = node[p]
+        if node is None or parts[-1] in node:
+            out[key] = event[key]
+        else:
+            node[parts[-1]] = event[key]
+    return out
+
+
+def start_search(alert):
+    # The alert's events are the ones tagged with its short id. Sekoia's own
+    # XSOAR integration searches from first_seen_at less a minute up to now:
+    # events are indexed slightly after the time they carry.
+    try:
+        first = datetime.fromisoformat(str(alert.get("first_seen_at")).replace("Z", "+00:00"))
+        earliest = (first - timedelta(minutes=1)).astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+    except ValueError:
+        earliest = "-7d"
+    job = request("POST", "/v1/sic/conf/events/search/jobs", body={
+        "earliest_time": earliest,
+        "latest_time": "now",
+        "term": "alert_short_ids:" + str(alert.get("short_id")),
+    })
+    return job.get("uuid")
+
+
+def collect(job_uuid, max_events):
+    path = "/v1/sic/conf/events/search/jobs/" + urllib.parse.quote(job_uuid, safe="")
+    while left() > 2:
+        if request("GET", path).get("status") == 2:
+            res = request("GET", path + "/events", params={"limit": max_events})
+            return [nest(e) for e in res.get("items") or []], res.get("total")
+        time.sleep(1)
+    return None, None
+
+
+def attach_events(alerts, max_events):
+    """Put each alert's events on it, as events (nested) and events_total. A
+    search not done within the budget leaves events_pending and the job uuid
+    (events_results_query reads it later): the alert still goes out, because
+    the rule's watermark moves on with this run and would not offer it again."""
+    def one(alert):
+        try:
+            job = start_search(alert)
+            events, total = collect(job, max_events) if job else (None, None)
+        except Exception as e:
+            alert["events_error"] = str(e)
+            return
+        if events is None:
+            alert["events_pending"] = True
+            alert["events_job_uuid"] = job
+        else:
+            alert["events"] = events
+            alert["events_total"] = total
+    with ThreadPoolExecutor(max_workers=10) as pool:
+        list(pool.map(one, alerts))
+
+
 def _range(value):
     # A bare date is the lower bound ("<date>,now"); a full range is passed through.
     v = str(value)
@@ -50,7 +132,10 @@ def run():
         params["range[urgency]"] = inp["urgency"]
     if inp.get("alerts_type"):
         params["match[type_value]"] = inp["alerts_type"]
-    print(json.dumps(request("GET", "/v1/sic/alerts", params=params)))
+    out = request("GET", "/v1/sic/alerts", params=params)
+    if str(inp.get("with_events") or "").lower() in ("true", "1", "yes"):
+        attach_events(out.get("items") or [], int(inp.get("max_events") or 10))
+    print(json.dumps(out))
 
 
 try: