feat(sentinelone): investigate_alert, one call for an agent (1.19.0)
Wraith spent ten calls on a real unified alert and still lost the host: it took the UAM asset id for an agent id and the raw agent list ran past what a tool result carries. investigate_alert reads the alert, its detail, the host's agent (by id or UUID, then name), the other alerts and threats on that host around the detection, and the file SHA1's reputation, blocklist and fleet-wide threats, summarised in one answer. The operator sheet now starts with it (budget 8). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015RhL2gy6Sx6Y3FEsyBE3mZ
This commit is contained in:
Claude
@@ -0,0 +1,409 @@
|
||||
import datetime, json, os, re, sys, urllib.request, urllib.parse, urllib.error
|
||||
|
||||
# One call for an agent investigating a SentinelOne alert: what the alert is,
|
||||
# the host behind it, and what else the console holds around it. An operator
|
||||
# model spent ten calls rebuilding this and still lost the host — it took the
|
||||
# UAM asset id (a 26-character token) for an agent id, and the raw agent list
|
||||
# it fell back to ran past what a tool result can carry. Every step here is
|
||||
# best effort and summarised: a step that fails is named in "errors" and the
|
||||
# rest of the answer still comes back.
|
||||
|
||||
UUID_RE = re.compile(r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$")
|
||||
SHA1_RE = re.compile(r"^[0-9a-fA-F]{40}$")
|
||||
|
||||
# List-shape fields (the ones get_unified_alerts reads, known to every console)
|
||||
# for the alert itself, fetched by an id filter.
|
||||
ALERT_FIELDS = """
|
||||
id name description severity status result analystVerdict classification
|
||||
confidenceLevel detectedAt firstSeenAt lastSeenAt storylineId fileName fileHash
|
||||
analytics { name category }
|
||||
detectionSource { product vendor engine }
|
||||
asset { id name agentUuid agentVersion osType osVersion lastLoggedInUser connectivityToConsole }
|
||||
detectionTime {
|
||||
asset { domain ipV4 lastLoggedInUser osName }
|
||||
attacker { host ip }
|
||||
scope { accountName groupName siteName }
|
||||
targetUser { name domain }
|
||||
}
|
||||
process { cmdLine parentName file { certSubject md5 name path sha1 sha256 } }
|
||||
"""
|
||||
|
||||
# What only the per-alert view carries.
|
||||
DETAIL_FIELDS = """
|
||||
indicators { message attacks { tactic { name } technique { name uid } } }
|
||||
observables { name value typeName }
|
||||
relatedAlerts { id name }
|
||||
process { username }
|
||||
"""
|
||||
|
||||
SIBLING_FIELDS = "id name severity status detectedAt classification"
|
||||
|
||||
|
||||
def request(method, url, headers, body=None):
|
||||
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=60) as resp:
|
||||
raw = resp.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def trim(v, n):
|
||||
s = "" if v is None else str(v)
|
||||
return s if len(s) <= n else s[:n] + "…"
|
||||
|
||||
|
||||
def compact(d):
|
||||
"""Drop the empty values, so the answer says only what the console said."""
|
||||
if isinstance(d, dict):
|
||||
out = {k: compact(v) for k, v in d.items()}
|
||||
return {k: v for k, v in out.items() if v not in (None, "", [], {})}
|
||||
if isinstance(d, list):
|
||||
return [compact(x) for x in d if x not in (None, "", [], {})]
|
||||
return d
|
||||
|
||||
|
||||
def parse_time(s):
|
||||
try:
|
||||
dt = datetime.datetime.fromisoformat(str(s).replace("Z", "+00:00"))
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return dt if dt.tzinfo else dt.replace(tzinfo=datetime.timezone.utc)
|
||||
|
||||
|
||||
def iso(dt):
|
||||
return dt.astimezone(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z")
|
||||
|
||||
|
||||
def group_count(items, key, extra=None, top=10):
|
||||
"""Count items per key value, most frequent first, with the latest time seen."""
|
||||
groups = {}
|
||||
for it in items:
|
||||
k = key(it) or "?"
|
||||
g = groups.setdefault(k, {"value": k, "count": 0, "last": ""})
|
||||
g["count"] += 1
|
||||
t = extra(it) if extra else ""
|
||||
if t and t > g["last"]:
|
||||
g["last"] = t
|
||||
out = sorted(groups.values(), key=lambda g: -g["count"])[:top]
|
||||
if not extra:
|
||||
for g in out:
|
||||
g.pop("last")
|
||||
return out
|
||||
|
||||
|
||||
class Console:
|
||||
def __init__(self, base, headers):
|
||||
self.base, self.headers, self.errors = base, headers, []
|
||||
|
||||
def get(self, step, path, qs=None, missing=None):
|
||||
"""missing, when given, is what a 404 means for this step: an answer,
|
||||
not a failed check."""
|
||||
url = self.base + path + ("?" + urllib.parse.urlencode({k: v for k, v in (qs or {}).items() if v not in (None, "")}) if qs else "")
|
||||
try:
|
||||
return request("GET", url, self.headers)
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code == 404 and missing is not None:
|
||||
return missing
|
||||
self.errors.append({"step": step, "error": "HTTP %d" % e.code,
|
||||
"detail": trim(e.read().decode("utf-8", "replace"), 300)})
|
||||
except Exception as e:
|
||||
self.errors.append({"step": step, "error": trim(e, 300)})
|
||||
return None
|
||||
|
||||
def gql(self, step, query):
|
||||
try:
|
||||
resp = request("POST", self.base + "/unifiedalerts/graphql", self.headers, {"query": query})
|
||||
except urllib.error.HTTPError as e:
|
||||
self.errors.append({"step": step, "error": "HTTP %d" % e.code,
|
||||
"detail": trim(e.read().decode("utf-8", "replace"), 300)})
|
||||
return None
|
||||
except Exception as e:
|
||||
self.errors.append({"step": step, "error": trim(e, 300)})
|
||||
return None
|
||||
if resp.get("errors"):
|
||||
self.errors.append({"step": step, "error": "GraphQL error",
|
||||
"detail": trim(json.dumps(resp["errors"]), 300)})
|
||||
return resp.get("data") or {}
|
||||
|
||||
|
||||
def alerts_query(filters, fields, first):
|
||||
# Both spellings of the sort argument exist across console versions; none
|
||||
# is needed here, so the query carries no sort at all.
|
||||
return """query Alerts { alerts(first: %d viewType: ALL filters: [ %s ]) {
|
||||
totalCount edges { node { %s } } } }""" % (first, " ".join(filters), fields)
|
||||
|
||||
|
||||
def summarize_agent(a):
|
||||
ips = []
|
||||
for nic in a.get("networkInterfaces") or []:
|
||||
for ip in nic.get("inet") or []:
|
||||
if ip and ip not in ips:
|
||||
ips.append(ip)
|
||||
return compact({
|
||||
"computer_name": a.get("computerName"),
|
||||
"agent_id": a.get("id"),
|
||||
"os": a.get("osName"),
|
||||
"os_revision": a.get("osRevision"),
|
||||
"machine_type": a.get("machineType"),
|
||||
"domain": a.get("domain"),
|
||||
"ips": ips[:6],
|
||||
"external_ip": a.get("externalIp"),
|
||||
"last_logged_in_user": a.get("lastLoggedInUserName"),
|
||||
"agent_version": a.get("agentVersion"),
|
||||
"is_active": a.get("isActive"),
|
||||
"network_status": a.get("networkStatus"),
|
||||
"infected": a.get("infected"),
|
||||
"active_threats": a.get("activeThreats"),
|
||||
"last_active": a.get("lastActiveDate"),
|
||||
"site": a.get("siteName"),
|
||||
"group": a.get("groupName"),
|
||||
"account": a.get("accountName"),
|
||||
})
|
||||
|
||||
|
||||
def find_agent(con, agent_uuid=None, agent_id=None, name=None):
|
||||
"""The host's agent, by the most precise key the alert gave."""
|
||||
for step, qs in (("agent_by_id", {"ids": agent_id}),
|
||||
("agent_by_uuid", {"uuids": agent_uuid}),
|
||||
("agent_by_name", {"computerName": name})):
|
||||
if not list(qs.values())[0]:
|
||||
continue
|
||||
resp = con.get(step, "/agents", dict(qs, limit=5))
|
||||
data = (resp or {}).get("data") or []
|
||||
if data:
|
||||
return data[0], step
|
||||
return None, None
|
||||
|
||||
|
||||
def threats_summary(con, step, qs):
|
||||
resp = con.get(step, "/threats", dict(qs, limit=100, sortBy="createdAt", sortOrder="desc"))
|
||||
if resp is None:
|
||||
return None
|
||||
data = resp.get("data") or []
|
||||
total = ((resp.get("pagination") or {}).get("totalItems"))
|
||||
return compact({
|
||||
"total": total if total is not None else len(data),
|
||||
"by_name": group_count(data, lambda t: (t.get("threatInfo") or {}).get("threatName"),
|
||||
lambda t: (t.get("threatInfo") or {}).get("createdAt") or ""),
|
||||
"by_host": group_count(data, lambda t: (t.get("agentRealtimeInfo") or {}).get("agentComputerName")),
|
||||
})
|
||||
|
||||
|
||||
def hash_checks(con, sha1, site_id):
|
||||
out = {"sha1": sha1}
|
||||
verdict = con.get("hash_verdict", "/hashes/%s/verdict" % sha1,
|
||||
missing={"data": {"verdict": "unknown to SentinelOne (404)"}})
|
||||
if verdict is not None:
|
||||
out["reputation"] = ((verdict.get("data") or {}).get("verdict")) or "unknown"
|
||||
blocked = []
|
||||
for scope_qs in ({"tenant": "true"}, {"siteIds": site_id} if site_id else None):
|
||||
if not scope_qs:
|
||||
continue
|
||||
resp = con.get("blocklist", "/restrictions", dict(scope_qs, type="black_hash", value__contains=sha1, limit=10))
|
||||
for r in (resp or {}).get("data") or []:
|
||||
blocked.append(compact({"scope": r.get("scopeName") or r.get("scope"), "description": trim(r.get("description"), 120)}))
|
||||
out["blocklisted"] = bool(blocked)
|
||||
if blocked:
|
||||
out["blocklist_entries"] = blocked[:5]
|
||||
out["threats_fleet_wide"] = threats_summary(con, "threats_by_hash", {"contentHashes": sha1})
|
||||
return compact(out)
|
||||
|
||||
|
||||
def investigate_uam(con, alert_id, days):
|
||||
data = con.gql("alert", alerts_query(['{ fieldId: "id" stringEqual: { value: %s } }' % json.dumps(alert_id)],
|
||||
ALERT_FIELDS, 1))
|
||||
if data is None:
|
||||
# The console could not be asked: that is a failed call, not an alert
|
||||
# that does not exist.
|
||||
raise RuntimeError("the alert could not be read: %s" % json.dumps(con.errors[-1]))
|
||||
edges = (data.get("alerts") or {}).get("edges") or []
|
||||
if not edges:
|
||||
return {"found": False, "alert_id": alert_id,
|
||||
"hint": "No unified alert has this id. Copy the id whole from get_incident_context (raw id), or pass a threat_id for a threat."}
|
||||
a = edges[0]["node"]
|
||||
det = (con.gql("alert_detail", "query D { alert(id: %s) { %s } }" % (json.dumps(alert_id), DETAIL_FIELDS)) or {}).get("alert") or {}
|
||||
|
||||
proc = a.get("process") or {}
|
||||
f = proc.get("file") or {}
|
||||
dt_asset = (a.get("detectionTime") or {}).get("asset") or {}
|
||||
scope = (a.get("detectionTime") or {}).get("scope") or {}
|
||||
indicators = []
|
||||
for ind in (det.get("indicators") or [])[:10]:
|
||||
attacks = ["%s / %s (%s)" % ((x.get("tactic") or {}).get("name"), (x.get("technique") or {}).get("name"),
|
||||
(x.get("technique") or {}).get("uid")) for x in ind.get("attacks") or []]
|
||||
indicators.append(compact({"message": trim(ind.get("message"), 220), "mitre": attacks[:3]}))
|
||||
observables, seen = [], set()
|
||||
for ob in (det.get("observables") or []):
|
||||
key = (ob.get("typeName") or ob.get("name"), ob.get("value"))
|
||||
if key in seen or not ob.get("value"):
|
||||
continue
|
||||
seen.add(key)
|
||||
observables.append("%s: %s" % (key[0], trim(key[1], 160)))
|
||||
out = {
|
||||
"kind": "unified alert",
|
||||
"alert": compact({
|
||||
"id": a.get("id"), "name": a.get("name"), "description": trim(a.get("description"), 400),
|
||||
"severity": a.get("severity"), "classification": a.get("classification"),
|
||||
"confidence": a.get("confidenceLevel"), "status": a.get("status"),
|
||||
"analyst_verdict": a.get("analystVerdict"), "mitigation_result": a.get("result"),
|
||||
"detected_at": a.get("detectedAt"), "first_seen": a.get("firstSeenAt"), "last_seen": a.get("lastSeenAt"),
|
||||
"rule": (a.get("analytics") or {}).get("name"), "rule_category": (a.get("analytics") or {}).get("category"),
|
||||
"detected_by": " ".join(x for x in ((a.get("detectionSource") or {}).get("product"),
|
||||
(a.get("detectionSource") or {}).get("engine")) if x),
|
||||
"storyline_id": a.get("storylineId"),
|
||||
"attacker": (a.get("detectionTime") or {}).get("attacker"),
|
||||
"target_user": (a.get("detectionTime") or {}).get("targetUser"),
|
||||
"scope": " / ".join(x for x in (scope.get("accountName"), scope.get("siteName"), scope.get("groupName")) if x),
|
||||
}),
|
||||
"process": compact({
|
||||
"command_line": trim(proc.get("cmdLine"), 600), "parent": proc.get("parentName"),
|
||||
"user": det.get("process", {}).get("username") if det.get("process") else None,
|
||||
"file": f.get("name") or a.get("fileName"), "path": f.get("path"),
|
||||
"sha1": f.get("sha1"), "sha256": f.get("sha256"), "md5": f.get("md5"),
|
||||
"signer": f.get("certSubject"),
|
||||
}),
|
||||
"indicators": indicators,
|
||||
"observables": observables[:25],
|
||||
"related_alerts": [trim(r.get("name"), 120) for r in (det.get("relatedAlerts") or [])[:10]],
|
||||
}
|
||||
|
||||
asset = a.get("asset") or {}
|
||||
agent, how = find_agent(con, agent_uuid=asset.get("agentUuid"), name=asset.get("name"))
|
||||
out["host"] = summarize_agent(agent) if agent else compact({
|
||||
"computer_name": asset.get("name"), "os": dt_asset.get("osName") or asset.get("osVersion"),
|
||||
"ip_at_detection": dt_asset.get("ipV4"), "domain": dt_asset.get("domain"),
|
||||
"last_logged_in_user": dt_asset.get("lastLoggedInUser") or asset.get("lastLoggedInUser"),
|
||||
"agent_version": asset.get("agentVersion"), "connectivity": asset.get("connectivityToConsole"),
|
||||
"note": "no SentinelOne agent found for this asset (not an endpoint, or uninstalled): fields from the alert",
|
||||
})
|
||||
if agent:
|
||||
out["host"]["found_by"] = how
|
||||
if dt_asset.get("ipV4"):
|
||||
out["host"]["ip_at_detection"] = dt_asset.get("ipV4")
|
||||
|
||||
anchor = parse_time(a.get("detectedAt")) or datetime.datetime.now(datetime.timezone.utc)
|
||||
start, end = anchor - datetime.timedelta(days=days), anchor + datetime.timedelta(days=1)
|
||||
window = "%s to %s (%d days before the alert's detection to 1 day after)" % (iso(start), iso(end), days)
|
||||
elsewhere = {"window": window}
|
||||
if asset.get("id"):
|
||||
sib = con.gql("same_host_alerts", alerts_query([
|
||||
'{ fieldId: "asset.id" stringEqual: { value: %s } }' % json.dumps(asset["id"]),
|
||||
'{ fieldId: "detectedAt" dateTimeRange: { start: %d end: %d } }' % (start.timestamp() * 1000, end.timestamp() * 1000),
|
||||
], SIBLING_FIELDS, 100))
|
||||
alerts = (sib or {}).get("alerts") or {}
|
||||
nodes = [e["node"] for e in alerts.get("edges") or [] if e.get("node") and e["node"].get("id") != alert_id]
|
||||
if sib is not None:
|
||||
elsewhere["same_host_other_alerts"] = compact({
|
||||
"total": max((alerts.get("totalCount") or 0) - 1, len(nodes)),
|
||||
"read": len(nodes),
|
||||
"by_name": group_count(nodes, lambda n: n.get("name"), lambda n: n.get("detectedAt") or ""),
|
||||
})
|
||||
if agent:
|
||||
elsewhere["same_host_threats"] = threats_summary(con, "threats_on_host", {
|
||||
"agentIds": agent.get("id"), "createdAt__gt": iso(start), "createdAt__lt": iso(end)})
|
||||
sha1 = f.get("sha1") or (a.get("fileHash") if SHA1_RE.match(str(a.get("fileHash") or "")) else None)
|
||||
if sha1:
|
||||
out["file_checks"] = hash_checks(con, sha1, agent.get("siteId") if agent else None)
|
||||
else:
|
||||
out["file_checks"] = "no SHA1 on this alert: reputation, blocklist and fleet-wide checks skipped"
|
||||
out["elsewhere"] = compact(elsewhere)
|
||||
return out
|
||||
|
||||
|
||||
def investigate_threat(con, threat_id, days):
|
||||
resp = con.get("threat", "/threats", {"ids": threat_id})
|
||||
if resp is None:
|
||||
raise RuntimeError("the threat could not be read: %s" % json.dumps(con.errors[-1]))
|
||||
data = (resp or {}).get("data") or []
|
||||
if not data:
|
||||
return {"found": False, "threat_id": threat_id,
|
||||
"hint": "No threat has this id. Copy the id whole from get_incident_context (threatInfo.threatId), or pass alert_id for a unified alert."}
|
||||
t = data[0]
|
||||
ti, rt, di = t.get("threatInfo") or {}, t.get("agentRealtimeInfo") or {}, t.get("agentDetectionInfo") or {}
|
||||
out = {
|
||||
"kind": "threat",
|
||||
"threat": compact({
|
||||
"id": ti.get("threatId"), "name": ti.get("threatName"), "classification": ti.get("classification"),
|
||||
"confidence": ti.get("confidenceLevel"), "analyst_verdict": ti.get("analystVerdict"),
|
||||
"incident_status": ti.get("incidentStatus"), "mitigation_status": ti.get("mitigationStatus"),
|
||||
"detection_type": ti.get("detectionType"), "engines": ti.get("engines"),
|
||||
"created_at": ti.get("createdAt"), "identified_at": ti.get("identifiedAt"),
|
||||
"storyline": ti.get("storyline"),
|
||||
}),
|
||||
"process": compact({
|
||||
"file_path": ti.get("filePath"), "sha1": ti.get("sha1"), "sha256": ti.get("sha256"), "md5": ti.get("md5"),
|
||||
"signer": ti.get("publisherName"), "verified": ti.get("fileVerificationType"),
|
||||
"originating_process": ti.get("originatorProcess"), "process_user": ti.get("processUser"),
|
||||
"command_line": trim(ti.get("maliciousProcessArguments"), 600),
|
||||
"initiated_by": ti.get("initiatedBy"),
|
||||
}),
|
||||
}
|
||||
notes = con.get("threat_notes", "/threats/%s/notes" % threat_id, {"limit": 10})
|
||||
out["notes"] = [compact({"by": n.get("creator"), "at": n.get("createdAt"), "text": trim(n.get("text"), 300)})
|
||||
for n in (notes or {}).get("data") or []]
|
||||
acts = con.get("threat_timeline", "/activities", {"threatIds": threat_id, "limit": 30, "sortBy": "createdAt", "sortOrder": "asc"})
|
||||
out["timeline"] = ["%s %s" % (x.get("createdAt"), trim(x.get("primaryDescription"), 200))
|
||||
for x in (acts or {}).get("data") or []][:20]
|
||||
|
||||
agent, how = find_agent(con, agent_id=rt.get("agentId"), name=rt.get("agentComputerName"))
|
||||
out["host"] = summarize_agent(agent) if agent else compact({
|
||||
"computer_name": rt.get("agentComputerName"), "os": rt.get("agentOsName"),
|
||||
"agent_version": rt.get("agentVersion"), "network_status": rt.get("networkStatus"),
|
||||
"note": "agent not found in the console: fields from the threat",
|
||||
})
|
||||
if agent:
|
||||
out["host"]["found_by"] = how
|
||||
out["host"]["user_at_detection"] = di.get("agentLastLoggedInUserName")
|
||||
out["host"]["ip_at_detection"] = di.get("agentIpV4")
|
||||
out["host"] = compact(out["host"])
|
||||
|
||||
anchor = parse_time(ti.get("createdAt")) or datetime.datetime.now(datetime.timezone.utc)
|
||||
start, end = anchor - datetime.timedelta(days=days), anchor + datetime.timedelta(days=1)
|
||||
elsewhere = {"window": "%s to %s (%d days before the threat to 1 day after)" % (iso(start), iso(end), days)}
|
||||
if agent:
|
||||
elsewhere["same_host_threats"] = threats_summary(con, "threats_on_host", {
|
||||
"agentIds": agent.get("id"), "createdAt__gt": iso(start), "createdAt__lt": iso(end)})
|
||||
out["elsewhere"] = compact(elsewhere)
|
||||
if ti.get("sha1"):
|
||||
out["file_checks"] = hash_checks(con, ti["sha1"], agent.get("siteId") if agent else None)
|
||||
return out
|
||||
|
||||
|
||||
def main():
|
||||
secrets = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
base = secrets.get("url", "").rstrip("/") + "/web/api/v2.1"
|
||||
con = Console(base, {
|
||||
"Authorization": "ApiToken " + secrets.get("api_token", ""),
|
||||
"Accept": "application/json",
|
||||
"Content-Type": "application/json",
|
||||
})
|
||||
alert_id = str(inputs.get("alert_id") or "").strip()
|
||||
threat_id = str(inputs.get("threat_id") or "").strip()
|
||||
days = max(1, min(int(inputs.get("days_before") or 7), 30))
|
||||
# A long number in alert_id is a threat id given in the wrong field.
|
||||
if alert_id.isdigit() and not threat_id:
|
||||
threat_id, alert_id = alert_id, ""
|
||||
if threat_id:
|
||||
if not threat_id.isdigit():
|
||||
raise ValueError("threat_id is a long number (about 19 digits); for a UUID use alert_id")
|
||||
out = investigate_threat(con, threat_id, days)
|
||||
elif alert_id:
|
||||
if not UUID_RE.match(alert_id):
|
||||
raise ValueError("alert_id is the unified alert's UUID (8-4-4-4-12 hex), copied whole; got %r (asset.id and agentUuid are not alert ids)" % alert_id)
|
||||
out = investigate_uam(con, alert_id, days)
|
||||
else:
|
||||
raise ValueError("give alert_id (unified alert UUID) or threat_id (threat number)")
|
||||
if con.errors:
|
||||
out["errors"] = con.errors[:8]
|
||||
print(json.dumps(out, ensure_ascii=False, separators=(",", ":")))
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
Reference in New Issue
Block a user