f3nris
b26257aaeb
An "Multiple User Accounts Were Disabled" investigation came back with one entity — the actor — while the layout stayed nearly empty. Four separate reasons, and the last one is why the first three keep recurring. evidences was an OBJECT, not an array. JSONata returns a bare object when the path yields one item, so the key had one shape on an investigation with a single alert and another with two. Bracketed to force an array. Only the actor was mapped. An ad_admin rule names who acted AND who it was done to, and on this alert the disabled accounts are the point of it. Target account, target user and both domains now map to the OCSF Target User group, the actor to Actor — the distinction that lets the sentence be read in one line. The domain controller was nowhere. A Windows event puts it in source_json.computerName and nowhere else, and source_json was skipped whole because the Azure AD one runs to kilobytes. Its top-level SCALARS are now read — bounded, and where computerName, osVersion and callerIpAddress live — while its nested part stays skipped. source_asset was the string "unknown". Mapped as-is it becomes an asset named "unknown" in the entity index, shared by every incident that had no asset: a pivot that groups things by the absence of a fact. Placeholders are dropped, and a key whose value cleans away to nothing is removed rather than kept empty, since "" is not nil and would still be written to its OCSF field. Five entities now, from one: both accounts, both display names, the DC. And the reason this kept happening. The layout builder's field picker is a closed <select> over the OCSF catalogue in web/src/lib/ocsf.ts, so a path absent from it cannot be placed on a layout at all — which is what `detection` and `evidence` are, and what every unanticipated key will be. Mapping each one to a real OCSF field is right where a field exists, but rule_keys_of_interest is an open vocabulary: four spellings turned up in as many days. So the whole of it is also rendered as a Markdown table into custom:rapid7_summary — keys of interest first, since the rule itself says those are what matter, then the evidence fields. Declare a custom field of type markdown with that name, put it on the layout, and a new rule type is readable the day it appears instead of after another mapping round. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
610 lines
26 KiB
Python
610 lines
26 KiB
Python
import json, os, sys, urllib.request, urllib.parse, urllib.error
|
|
|
|
# The Alert Triage API lives under /idr/at, a base of its own — not /idr/v1 or
|
|
# /idr/v2 like the investigation endpoints. Same host, same API key.
|
|
AT = "idr/at/"
|
|
|
|
|
|
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"}
|
|
multi = str(s.get("is_multi_customer") or "").strip().lower() in ("1", "true", "yes")
|
|
return base, headers, multi
|
|
|
|
|
|
def _cfg_region():
|
|
return str(json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")).get("region") or "us").strip().lower()
|
|
|
|
|
|
def _inputs():
|
|
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
|
|
|
|
|
def _list(v):
|
|
if v in (None, ""):
|
|
return []
|
|
if isinstance(v, list):
|
|
return [str(x).strip() for x in v if str(x).strip()]
|
|
return [p.strip() for p in str(v).split(",") if p.strip()]
|
|
|
|
|
|
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 += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean, doseq=True)
|
|
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=90) as r:
|
|
raw = r.read()
|
|
return json.loads(raw) if raw else {}
|
|
|
|
import re
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
ISO = "%Y-%m-%dT%H:%M:%SZ"
|
|
RRN_BATCH = 100
|
|
DEFAULT_BUDGET_S = 60
|
|
|
|
# idr/v2/investigations answers with 13 fields, three of them null: a title, a
|
|
# priority, a status, an assignee and timestamps. No host, no account, no
|
|
# process, no MITRE code — and not one of the alerts the investigation
|
|
# aggregates. An incident built from it indexes zero entities, so it cannot be
|
|
# pivoted on, cannot be correlated, and gives an analyst nothing to read.
|
|
#
|
|
# The detection content is two hops away: the investigation's alert list gives
|
|
# RRNs, and the Alert Triage API turns those into alerts carrying
|
|
# rule_keys_of_interest (the host, the process, its command line, its parent,
|
|
# the hashes) and rule.mitre_tcodes. This command walks both hops and returns
|
|
# the investigation with its alerts inlined, so the INVESTIGATION stays the unit
|
|
# that becomes an incident — results_path = data, dedup on rrn, exactly as the
|
|
# plain investigation feed. Only the command changes.
|
|
|
|
|
|
def _alert_rrn(a):
|
|
"""The alert's RRN, or None when this alert has none.
|
|
|
|
The investigation alert list is not consistent about where the RRN sits.
|
|
Some entries carry it in `id` (rrn:alerts:eu:<org>:alert:1:<hex>); others put
|
|
a plain UUID there and the RRN, if any, in `rrn`. `rrn` is therefore read
|
|
first, and anything that is not RRN-shaped is dropped rather than sent on.
|
|
|
|
That last part is not defensive habit. /alerts/ops/rrns validates the whole
|
|
list and rejects the entire request when one entry is malformed, so a single
|
|
UUID slipping through costs every alert in the batch — which is exactly how
|
|
this failed: HTTP 400, eight invalid RRNs, zero alerts fetched, the whole run
|
|
lost. Dropping them here means the investigation still arrives, with its
|
|
unretrievable alerts counted rather than silently missing.
|
|
"""
|
|
for v in (a.get("rrn"), a.get("id")):
|
|
if isinstance(v, str) and v.startswith("rrn:"):
|
|
return v
|
|
return None
|
|
|
|
|
|
def _rrn_region(rrn):
|
|
"""The region segment of an RRN, or None.
|
|
|
|
An RRN is rrn:<service>:<region>:<org>:<type>:… — so a resource states which
|
|
region it lives in, while the base URL is built from the instance's
|
|
configured region. Those two agree on a single-region tenant and need not on
|
|
a multi-customer key, which spans organizations.
|
|
"""
|
|
parts = str(rrn or "").split(":")
|
|
return parts[2] if len(parts) > 4 and parts[0] == "rrn" and parts[2] else None
|
|
|
|
|
|
def _collect_region_failures(out, sink):
|
|
"""Carry a partial failure out of a response instead of dropping it.
|
|
|
|
The Alert Triage API answers HTTP 200 with a region_failures list when it
|
|
could not reach one of the regions behind the request — the spec says the
|
|
presence of items there IS the partial failure. Reading only the payload
|
|
turns that into missing alerts with no error anywhere, which for an
|
|
ingestion feed is indistinguishable from a quiet period.
|
|
"""
|
|
for f in (out.get("region_failures") or []):
|
|
sink.append(f)
|
|
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+)')
|
|
|
|
|
|
# Values a source writes when it has nothing to say. Mapped as-is they become an
|
|
# asset named "unknown" in the entity index, shared by every incident that had no
|
|
# asset — a pivot that groups things by the absence of a fact.
|
|
_PLACEHOLDERS = {"unknown", "n/a", "na", "none", "null", "-", ""}
|
|
|
|
|
|
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]
|
|
v = v.replace('\\"', '"').strip().strip('"')
|
|
return "" if v.strip().lower() in _PLACEHOLDERS else v
|
|
|
|
|
|
def _evidence_fields(ev):
|
|
"""One evidence's event as a flat, normalised dict.
|
|
|
|
`data` is a JSON-encoded STRING. JSONata cannot deserialise, so a mapper can
|
|
never reach into it; this is why the parsing happens here.
|
|
|
|
It arrives in at least two shapes and the union of both is read, rather than
|
|
one or the other. The first version of this handled only the wrapped shape
|
|
and returned {} for everything else — silently, so an Azure AD sign-in
|
|
carrying its account, its source address and its geolocation produced an
|
|
incident with no observable at all.
|
|
|
|
wrapped {source_data: "<syslog line>", custom_data: {Forti_user: …}}
|
|
An unparsed source. custom_data is the vendor's normalised view,
|
|
its keys prefixed per source; source_data is the raw line.
|
|
|
|
flat {user: …, account: …, source_ip: …, geoip_country_code: …,
|
|
source_json: {…}, r7_context: {…}, detection_context: {…}}
|
|
A parsed source. The event IS the object.
|
|
|
|
Scalars at the top level are always taken, so a third shape still yields its
|
|
plain fields instead of nothing. source_json is deliberately skipped — it is
|
|
the vendor's whole payload, several kilobytes of it, and it stays in the raw
|
|
line. r7_context and detection_context are flattened one level because they
|
|
are Rapid7's own annotations and are the same across parsed sources.
|
|
"""
|
|
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 = {}
|
|
|
|
# Flat shape: every scalar the event states about itself.
|
|
for k, v in blob.items():
|
|
if k in ("source_data", "custom_data", "source_json"):
|
|
continue
|
|
if isinstance(v, bool) or isinstance(v, (int, float)):
|
|
fields[k.lower()] = v
|
|
elif isinstance(v, str):
|
|
fields[k.lower()] = _clean(v)
|
|
elif isinstance(v, dict) and k in ("r7_context", "detection_context"):
|
|
for sk, sv in v.items():
|
|
if isinstance(sv, dict):
|
|
for ssk, ssv in sv.items():
|
|
if not isinstance(ssv, (str, int, float)) or isinstance(ssv, bool):
|
|
continue
|
|
# detection_context nests user.user_age_mins, so a blind
|
|
# join yields user_user_age_mins.
|
|
name = ssk if ssk.startswith(sk + "_") else sk + "_" + ssk
|
|
fields[name.lower()] = _clean(ssv) if isinstance(ssv, str) else ssv
|
|
elif isinstance(sv, (str, int, float)):
|
|
fields[(k + "_" + sk).lower()] = _clean(sv) if isinstance(sv, str) else sv
|
|
|
|
# source_json is the vendor's own payload. Its NESTED part is skipped — the
|
|
# Azure AD one runs to several kilobytes — but its top-level scalars are
|
|
# small, bounded and often the only place something essential appears: a
|
|
# Windows 4725 puts the domain controller in computerName and nowhere else,
|
|
# and an Azure sign-in puts the caller address in callerIpAddress. Existing
|
|
# keys win, so the event's own naming is never overwritten by the vendor's.
|
|
sj = blob.get("source_json")
|
|
if isinstance(sj, dict):
|
|
for k, v in sj.items():
|
|
key = k.lower()
|
|
if key in fields:
|
|
continue
|
|
if isinstance(v, bool) or isinstance(v, (int, float)):
|
|
fields[key] = v
|
|
elif isinstance(v, str) and v.strip():
|
|
fields[key] = _clean(v)
|
|
|
|
# Wrapped shape, overlaid on top when present.
|
|
custom = blob.get("custom_data")
|
|
if isinstance(custom, dict):
|
|
for k, v in custom.items():
|
|
if not isinstance(v, (str, int, float)) or isinstance(v, bool):
|
|
continue
|
|
key = k.split("_", 1)[1] if "_" in k else k
|
|
fields[key.lower()] = _clean(v) if isinstance(v, str) else v
|
|
source = blob.get("source_data")
|
|
if isinstance(source, str):
|
|
for k, v in _KV.findall(source):
|
|
fields[k.lower()] = _clean(v)
|
|
|
|
# A key whose value cleaned away to nothing is dropped rather than kept
|
|
# empty: an empty string is not nil, so it would still be written to its OCSF
|
|
# field, and device.hostname = "" asserts an asset with no name.
|
|
fields = {k: v for k, v in fields.items() if v != ""}
|
|
return fields, (source if isinstance(source, str) else raw)
|
|
|
|
|
|
# Event types worth asking a process tree for. The endpoint sensor produces one;
|
|
# an authentication or firewall alert never will, and on a feed where those
|
|
# dominate, asking anyway is one wasted call per alert. include_process_tree=true
|
|
# overrides this and asks for every alert.
|
|
_PROCESS_EVENT_TYPES = {
|
|
"process_start_event", "advanced_malware", "virus",
|
|
"endpoint_activity", "file_modification", "local_service_installed",
|
|
}
|
|
|
|
# Guards on a tree of unknown shape: a wide one would otherwise expand into more
|
|
# chains than anybody reads, and a cycle would not terminate at all.
|
|
_MAX_CHAINS = 12
|
|
_MAX_DEPTH = 24
|
|
|
|
|
|
def _flatten_tree(root, out, path=None, depth=0):
|
|
"""Rapid7 nests a process tree through `children`; the layout block reads a
|
|
flat chain, root first. So each root-to-leaf path becomes one chain.
|
|
|
|
The block marks the LAST node of a chain as the target. Rapid7 flags no node
|
|
as the one the alert fired on, so that mark means "end of this branch" here
|
|
and nothing stronger — inventing a target would assert something the sensor
|
|
never said.
|
|
"""
|
|
if not isinstance(root, dict) or depth > _MAX_DEPTH or len(out) >= _MAX_CHAINS:
|
|
return
|
|
name = root.get("name")
|
|
node = {}
|
|
if name:
|
|
node["name"] = str(name)
|
|
if root.get("cmdline"):
|
|
node["cmd_line"] = str(root["cmdline"])
|
|
pid = root.get("pid")
|
|
if pid not in (None, ""):
|
|
try:
|
|
node["pid"] = int(str(pid))
|
|
except ValueError:
|
|
pass
|
|
h = root.get("hashes") or {}
|
|
if isinstance(h, dict):
|
|
for k in ("md5", "sha256"):
|
|
if h.get(k):
|
|
node[k] = str(h[k])
|
|
if root.get("start_time") not in (None, ""):
|
|
node["started_at"] = root["start_time"]
|
|
if root.get("username"):
|
|
node["_username"] = str(root["username"])
|
|
chain = (path or []) + ([node] if node else [])
|
|
children = [c for c in (root.get("children") or []) if isinstance(c, dict)]
|
|
if not children:
|
|
if chain:
|
|
out.append(chain)
|
|
return
|
|
for c in children:
|
|
_flatten_tree(c, out, chain, depth + 1)
|
|
|
|
|
|
def _tree_chains(payload, alert):
|
|
"""The process_tree block's own shape: N chains, each with its nodes."""
|
|
chains = []
|
|
for tree in (payload.get("process_trees") or []):
|
|
if not isinstance(tree, dict):
|
|
continue
|
|
paths = []
|
|
_flatten_tree(tree.get("process_tree"), paths)
|
|
for nodes in paths:
|
|
user = ""
|
|
for n in nodes:
|
|
user = n.pop("_username", user) or user
|
|
if not nodes:
|
|
continue
|
|
chain = {"nodes": nodes, "alert_id": tree.get("alert_rrn") or alert.get("rrn")}
|
|
if user:
|
|
chain["user"] = user
|
|
if alert.get("type"):
|
|
chain["rule"] = alert["type"]
|
|
host = (alert.get("evidence_fields") or {}).get("devname") or (alert.get("evidence_fields") or {}).get("hostname")
|
|
if not host:
|
|
for kv in (alert.get("rule_keys_of_interest") or []):
|
|
if kv.get("key") == "hostname" and kv.get("values"):
|
|
host = kv["values"][0]
|
|
if host:
|
|
chain["host"] = host
|
|
chains.append(chain)
|
|
if len(chains) >= _MAX_CHAINS:
|
|
return chains
|
|
return chains
|
|
|
|
|
|
# Keys already rendered elsewhere in the incident, or too noisy for a summary.
|
|
_MD_SKIP = {"entry_id", "timestamp"}
|
|
_MD_MAX_ROWS = 40
|
|
_MD_MAX_VALUE = 300
|
|
|
|
|
|
def _markdown_summary(inv, alerts):
|
|
"""A readable rendering of what the detection actually says.
|
|
|
|
The layout builder's field picker is a closed <select> over the OCSF
|
|
catalogue, so a path that is not in it — `detection`, `evidence`, anything
|
|
an unanticipated rule emits — cannot be placed on a layout at all. Mapping
|
|
each key to a real OCSF field is the right answer where one exists, but
|
|
rule_keys_of_interest is an open vocabulary and always will be: three
|
|
vocabularies turned up in as many days.
|
|
|
|
So the whole of it is also rendered as Markdown and mapped to a custom
|
|
field, which a layout CAN show. Nothing is lost to the analyst even when
|
|
nothing was mapped, and the mapping work stops being what stands between
|
|
a new rule type and a readable incident.
|
|
"""
|
|
lines = []
|
|
for a in alerts:
|
|
title = a.get("type") or a.get("title") or "Alert"
|
|
lines.append("### " + str(title))
|
|
rule = a.get("rule") or {}
|
|
tcodes = [t for t in (rule.get("mitre_tcodes") or []) if t]
|
|
meta = []
|
|
if a.get("triggering_event_type"):
|
|
meta.append("`" + str(a["triggering_event_type"]) + "`")
|
|
if tcodes:
|
|
meta.append("ATT&CK " + ", ".join(str(t) for t in tcodes))
|
|
if a.get("status"):
|
|
meta.append("status " + str(a["status"]))
|
|
if a.get("disposition"):
|
|
meta.append("disposition " + str(a["disposition"]))
|
|
if meta:
|
|
lines.append(" · ".join(meta))
|
|
rows = []
|
|
seen = set()
|
|
# Keys of interest first: the rule itself says these are what matter.
|
|
for kv in (a.get("rule_keys_of_interest") or []):
|
|
k = kv.get("key")
|
|
vals = [str(v) for v in (kv.get("values") or []) if v not in (None, "")]
|
|
if k and vals and k not in seen:
|
|
seen.add(k)
|
|
rows.append((k, ", ".join(vals)))
|
|
for k, v in sorted((a.get("evidence_fields") or {}).items()):
|
|
if k in seen or k in _MD_SKIP or v in (None, ""):
|
|
continue
|
|
seen.add(k)
|
|
rows.append((k, str(v)))
|
|
if rows:
|
|
lines.append("")
|
|
lines.append("| Field | Value |")
|
|
lines.append("| --- | --- |")
|
|
for k, v in rows[:_MD_MAX_ROWS]:
|
|
if len(v) > _MD_MAX_VALUE:
|
|
v = v[:_MD_MAX_VALUE] + "…"
|
|
# A pipe inside a value would split the row into a new column.
|
|
lines.append("| " + k.replace("|", "\\|") + " | " + v.replace("|", "\\|") + " |")
|
|
if len(rows) > _MD_MAX_ROWS:
|
|
lines.append("")
|
|
lines.append("_%d more field(s) not shown._" % (len(rows) - _MD_MAX_ROWS))
|
|
lines.append("")
|
|
return "\n".join(lines).strip()
|
|
|
|
|
|
def run():
|
|
_, _, multi = _cfg()
|
|
inp = _inputs()
|
|
limit = int(inp.get("limit") or 50)
|
|
max_alerts = int(inp.get("max_alerts") or 500)
|
|
budget = float(inp.get("time_budget_seconds") or DEFAULT_BUDGET_S)
|
|
started = time.monotonic()
|
|
|
|
start_time = inp.get("start_time")
|
|
if not start_time:
|
|
days = int(inp.get("default_days") or 1)
|
|
start_time = (datetime.now(timezone.utc) - timedelta(days=days)).strftime(ISO)
|
|
|
|
inv_page = request("GET", "idr/v2/investigations", params={
|
|
"index": inp.get("index") or "0",
|
|
"size": inp.get("page_size") or limit,
|
|
"statuses": inp.get("statuses"),
|
|
"priorities": inp.get("priorities"),
|
|
"start_time": start_time,
|
|
"end_time": inp.get("end_time"),
|
|
"sort_field": "Created time",
|
|
"sort_direction": "ASC",
|
|
"multi-customer": "true" if multi else None,
|
|
})
|
|
investigations = (inv_page.get("data") or [])[:limit]
|
|
|
|
# Alert RRNs per investigation. One call each — there is no bulk form — so
|
|
# this is the step that grows with the window, and the one the time budget
|
|
# guards. An investigation whose alerts cannot be read keeps its place in the
|
|
# output with an alerts_error: losing the whole run over one of them, or
|
|
# dropping it silently, would both be worse.
|
|
rrns_by_inv, order, truncated, skipped = {}, [], False, 0
|
|
region_failures, regions_seen = [], set()
|
|
for inv in investigations:
|
|
inv_rrn = inv.get("rrn") or ""
|
|
if not inv_rrn:
|
|
continue
|
|
if time.monotonic() - started > budget or len(order) >= max_alerts:
|
|
truncated = True
|
|
break
|
|
try:
|
|
page = request("GET", "idr/v2/investigations/" + urllib.parse.quote(inv_rrn, safe="") + "/alerts",
|
|
params={"multi-customer": "true" if multi else None})
|
|
except urllib.error.HTTPError as e:
|
|
inv["alerts_error"] = "HTTP " + str(e.code)
|
|
continue
|
|
got = []
|
|
for a in (page.get("data") or []):
|
|
rrn = _alert_rrn(a)
|
|
if rrn is None:
|
|
skipped += 1
|
|
continue
|
|
got.append(rrn)
|
|
r = _rrn_region(rrn)
|
|
if r:
|
|
regions_seen.add(r)
|
|
if rrn not in order:
|
|
order.append(rrn)
|
|
rrns_by_inv[inv_rrn] = got
|
|
|
|
# The alerts themselves, batched, then indexed so each investigation can be
|
|
# handed its own.
|
|
field_ids = _list(inp.get("field_ids"))
|
|
by_rrn = {}
|
|
for i in range(0, len(order), RRN_BATCH):
|
|
body = {"rrns": order[i:i + RRN_BATCH]}
|
|
if field_ids:
|
|
body["field_ids"] = field_ids
|
|
out = _collect_region_failures(request("POST", AT + "alerts/ops/rrns", body=body), region_failures)
|
|
for a in (out.get("alerts") or []):
|
|
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")
|
|
|
|
# ⑤ Process trees. A third call per alert, so it is gated: "auto" asks only
|
|
# for the event types an endpoint sensor produces one for, "true" asks
|
|
# always, "false" never. Chains are attached to the alert AND to the
|
|
# investigation, because the layout block reads ctx.process_tree at the
|
|
# top level.
|
|
tree_mode = str(inp.get("include_process_tree") or "auto").strip().lower()
|
|
tree_calls, tree_errors = 0, 0
|
|
if tree_mode != "false":
|
|
for rrn, a in by_rrn.items():
|
|
if time.monotonic() - started > budget:
|
|
truncated = True
|
|
break
|
|
if tree_mode == "auto" and str(a.get("triggering_event_type") or "") not in _PROCESS_EVENT_TYPES:
|
|
continue
|
|
try:
|
|
page = request("POST", AT + "alerts/" + urllib.parse.quote(rrn, safe="") + "/process_trees/latest",
|
|
params={"index": "0", "size": "5"})
|
|
tree_calls += 1
|
|
except urllib.error.HTTPError as e:
|
|
# 404 is the normal answer for an alert that has no tree, not a
|
|
# fault: recorded only when it is something else.
|
|
if e.code != 404:
|
|
a["process_tree_error"] = "HTTP " + str(e.code)
|
|
tree_errors += 1
|
|
continue
|
|
chains = _tree_chains(page, a)
|
|
if chains:
|
|
a["process_tree"] = chains
|
|
|
|
fetched = 0
|
|
for inv in investigations:
|
|
rrns = rrns_by_inv.get(inv.get("rrn") or "", [])
|
|
alerts = [by_rrn[r] for r in rrns if r in by_rrn]
|
|
inv["alerts"] = alerts
|
|
inv["alert_count"] = len(rrns)
|
|
# The block reads ctx.process_tree at the top level and draws one chain
|
|
# per alert, so every alert's chains are gathered here rather than left
|
|
# nested where no mapper expression could reach past alerts[0].
|
|
if alerts:
|
|
inv["summary_markdown"] = _markdown_summary(inv, alerts)
|
|
rolled = []
|
|
for a in alerts:
|
|
rolled.extend(a.get("process_tree") or [])
|
|
if rolled:
|
|
inv["process_tree"] = rolled[:_MAX_CHAINS]
|
|
# Say it when the alert list is a sample: alert_count is what the
|
|
# investigation actually has, len(alerts) what this run retrieved.
|
|
if len(alerts) < len(rrns):
|
|
inv["alerts_truncated"] = True
|
|
fetched += len(alerts)
|
|
# Lift a few first-alert values to the top level. The incident-field
|
|
# mapper resolves dotted paths only — it cannot index an array — so
|
|
# `alerts[0].ai_suggested_disposition` in a field_mappings entry reads as
|
|
# nothing, silently, and the custom field arrives empty with no error
|
|
# anywhere. The OCSF entries are evaluated by full JSONata and do not
|
|
# need this; only field_mappings does.
|
|
if alerts:
|
|
a = alerts[0]
|
|
log = (a.get("log_details") or [{}])[0]
|
|
inv.update({
|
|
"first_alert_rrn": a.get("rrn"),
|
|
"first_alert_type": a.get("type"),
|
|
"first_alert_source": a.get("external_source"),
|
|
"first_alert_status": a.get("status"),
|
|
"first_alert_disposition": a.get("disposition"),
|
|
"first_alert_ai_disposition": a.get("ai_suggested_disposition"),
|
|
"first_alert_event_type": a.get("triggering_event_type"),
|
|
"first_alert_rule_rrn": (a.get("rule") or {}).get("rrn"),
|
|
"first_alert_log_id": log.get("log_id"),
|
|
"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,
|
|
"metadata": {
|
|
"investigations_returned": len(investigations),
|
|
"alert_rrns_collected": len(order),
|
|
"alerts_fetched": fetched,
|
|
"evidence_calls": evidence_calls,
|
|
"evidence_errors": evidence_errors,
|
|
"process_tree_calls": tree_calls,
|
|
"process_tree_errors": tree_errors,
|
|
"alert_ids_without_rrn": skipped,
|
|
"truncated": truncated,
|
|
# HTTP 200 with a non-empty region_failures is a PARTIAL result:
|
|
# alerts from that region are simply absent. Surfaced so a shrinking
|
|
# feed has a cause attached to it.
|
|
"region_failures": region_failures,
|
|
"regions_seen": sorted(regions_seen),
|
|
"configured_region": _cfg_region(),
|
|
"region_mismatch": sorted(r for r in regions_seen if r != _cfg_region()),
|
|
"elapsed_seconds": round(time.monotonic() - started, 1),
|
|
},
|
|
}))
|
|
|
|
|
|
try:
|
|
run()
|
|
except urllib.error.HTTPError as e:
|
|
detail = e.read().decode("utf-8", "replace")
|
|
if e.code in (401, 403):
|
|
detail += (" — the Alert Triage API (/idr/at) is only available to Managed Detection and"
|
|
" Response customers with access to the Alerts experience; a key that works on"
|
|
" the investigation endpoints can still be refused here")
|
|
print(json.dumps({"error": "HTTP " + str(e.code), "detail": detail}))
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(json.dumps({"error": str(e)}))
|
|
sys.exit(1)
|