diff --git a/integrations/sekoia-xdr/manifest.yaml b/integrations/sekoia-xdr/manifest.yaml index 7447f87..e5b4f0c 100644 --- a/integrations/sekoia-xdr/manifest.yaml +++ b/integrations/sekoia-xdr/manifest.yaml @@ -1,8 +1,8 @@ id: sekoia_xdr name: Sekoia XDR -version: 1.2.1 +version: 1.3.0 description: "Sekoia Defend (XDR) REST API — alert ingestion and lifecycle (list/get/search, status workflow, comments), event search jobs (create/status/results + one-shot search), cases, asset management (get/list, attributes/keys add/remove), users, kill chains and a generic authenticated HTTP passthrough." -changelog: "1.2.0 — Mapper reviewed on a real Entra ID alert. The incident source is now 'Sekoia XDR': it carried Sekoia's own `source` field, the alert's source address, which now goes to src_endpoint.ip (or .hostname for a host), and `target` to dst_endpoint instead of src_endpoint. The description is the alert's sentence with its {rule:} placeholder replaced by the rule name, rather than the rule documentation (now finding_info.analytic.desc). Every TTP is mapped into attacks, not only the first; kill_chain_short_id, Sekoia's own kill chain, no longer poses as an ATT&CK tactic uid. Times are epoch milliseconds as OCSF expects (created_at/updated_at came in seconds, first/last seen as text); alert category in finding_info.types, detection type in analytic.category, product in metadata. | 1.1.0 — Declare the risk tier on the 1 command that carried none, all of them destructive. An unset risk was already gated as destructive, so the approval prompt does not move; what changes is that the tier is a stated decision rather than a default, and a command added later cannot inherit it by omission. | 1.0.0 — Initial release: alert ingestion (list_alerts) with an exhaustive OCSF mapper, 20 commands across alerts, event search, cases, assets, users and kill chains." +changelog: "1.3.0 — list_alerts can attach each alert's events (with_events, max_events). A Sekoia alert only summarises its detection; the values an analyst acts on — addresses, user, process, command line, URL — are in the events tagged with its short id. Each alert gets one event search over alert_short_ids, from first_seen_at less a minute to now as Sekoia's own XSOAR integration does, run in parallel within the 60 s a command is given. The events come back with their dotted ECS names rebuilt into a tree, so a mapper reads events[0].source.ip. An alert whose search is not done in time still goes out, with events_pending and events_job_uuid, since the alert rule's watermark would not offer it again. | 1.2.0 — Mapper reviewed on a real Entra ID alert. The incident source is now 'Sekoia XDR': it carried Sekoia's own `source` field, the alert's source address, which now goes to src_endpoint.ip (or .hostname for a host), and `target` to dst_endpoint instead of src_endpoint. The description is the alert's sentence with its {rule:} placeholder replaced by the rule name, rather than the rule documentation (now finding_info.analytic.desc). Every TTP is mapped into attacks, not only the first; kill_chain_short_id, Sekoia's own kill chain, no longer poses as an ATT&CK tactic uid. Times are epoch milliseconds as OCSF expects (created_at/updated_at came in seconds, first/last seen as text); alert category in finding_info.types, detection type in analytic.category, product in metadata. | 1.1.0 — Declare the risk tier on the 1 command that carried none, all of them destructive. An unset risk was already gated as destructive, so the approval prompt does not move; what changes is that the tier is a stated decision rather than a default, and a command added later cannot inherit it by omission. | 1.0.0 — Initial release: alert ingestion (list_alerts) with an exhaustive OCSF mapper, 20 commands across alerts, event search, cases, assets, users and kill chains." category: siem # Per-instance configuration. The API host defaults to the EU region; an @@ -46,6 +46,8 @@ commands: sort_by: { type: string, description: "Field to sort by (created_at, updated_at, target, urgency, status). Default created_at." } direction: { type: string, description: "Sort direction (asc, desc). Default asc." } offset: { type: number, description: "Result offset for paging" } + with_events: { type: boolean, description: "Attach each alert's events (searched by its short_id, first_seen_at less a minute to now) as events[], ECS names nested (events[0].source.ip), with events_total. A search not finished in time leaves events_pending and events_job_uuid for events_results_query. Adds one event search per alert: keep limit around 20." } + max_events: { type: number, description: "Events kept per alert when with_events is set (default 10)." } required: [] outputs_schema: { properties: {} } ingest: diff --git a/integrations/sekoia-xdr/scripts/list_alerts.py b/integrations/sekoia-xdr/scripts/list_alerts.py index c01382c..820a039 100644 --- a/integrations/sekoia-xdr/scripts/list_alerts.py +++ b/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 (",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: