feat(sentinelone): fetch the unified (UAM) alerts (v1.8.0)
The package could already write to a unified alert — update its status and its analyst verdict — but never read one. The only fetch was get_alerts, which reads /cloud-detection/alerts: a different object, covering only the detections the STAR and custom rules raise. Everything SentinelOne surfaces through the console-wide alert model — cloud, identity, third-party — had no way into an incident. get_unified_alerts reads the unifiedalerts GraphQL endpoint, alongside the two mutations already pointed at it. Notes on the shape it returns: - edges/node is flattened to a plain data[]. The envelope carries no id of its own, so a rule dedup'ing on a top-level id would key every alert on nothing; flattened, the ingest block matches the REST commands in this package (results_path: data, dedup_key: id). - Sorted ascending, so a run truncated by the limit leaves the watermark on the oldest unseen alert instead of skipping everything behind the newest. - created_after is normalised from ISO8601, epoch seconds or epoch milliseconds: the scheduler injects the watermark raw, in whichever of the three formats the rule was configured with. The script builds the GraphQL filter itself, so the incremental input stays a plain timestamp. - A GraphQL error arrives as HTTP 200 with an errors array. Left unread, an unusable query would look like an endless run of empty, successful fetches. The mapper covers 74 fields. Its severity branch ends on the raw value rather than a number, so an alert with no severity — or a level SentinelOne adds later — falls through to the rule's default instead of being asserted Informational on data that never said so. A UAM alert also carries fields with no honest OCSF home (assignee, ticketId, the kubernetes controllers); they are fetched and left in the raw payload rather than forced into an unrelated field. The package default ingestion still points at get_threats: the new command is opt-in from the Collect editor, so no existing instance changes what it collects. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,184 @@
|
||||
import json, os, sys, urllib.request, urllib.parse, urllib.error
|
||||
|
||||
# Unified (UAM) alerts live behind a GraphQL endpoint, not the REST v2.1 tree the
|
||||
# rest of this package uses. The console-wide alert model is a different object
|
||||
# from /cloud-detection/alerts (see get_alerts): one alert per detection across
|
||||
# endpoint, cloud, identity and third-party surfaces.
|
||||
VIEW_TYPES = ("ALL", "CLOUD", "CUSTOM_ALERTS", "ENDPOINT", "IDENTITY", "STAR", "THIRD_PARTY")
|
||||
|
||||
# Fields requested per alert node. Kept in one place so the mapper and this
|
||||
# query cannot drift apart.
|
||||
NODE_FIELDS = """
|
||||
id name description severity status result analystVerdict attackSurfaces
|
||||
classification confidenceLevel externalId createdAt updatedAt firstSeenAt
|
||||
lastSeenAt detectedAt noteExists dataSources storylineId ticketId
|
||||
fileName fileHash
|
||||
analytics { uid name type typeValue category }
|
||||
assignee { userId fullName email }
|
||||
detectionSource { product vendor engine }
|
||||
asset {
|
||||
id name agentUuid agentVersion assetTypeClassifier category subcategory type
|
||||
connectivityToConsole osType osVersion pendingReboot lastLoggedInUser policy
|
||||
}
|
||||
detectionTime {
|
||||
asset {
|
||||
agentVersion consoleIpAddress domain ipV4 ipV6 lastLoggedInUser
|
||||
osName osRevision osType policy subscriptionTime
|
||||
}
|
||||
attacker { host ip }
|
||||
cloud { accountId cloudProvider image instanceId instanceSize location network tags }
|
||||
kubernetes {
|
||||
clusterName namespaceName nodeName podName containerId containerImageName
|
||||
containerName containerNetworkStatus controllerName controllerType
|
||||
}
|
||||
scope { accountId accountName groupName siteName }
|
||||
targetUser { name emailAddress domain }
|
||||
}
|
||||
process { cmdLine parentName file { certSubject md5 name path sha1 sha256 } }
|
||||
realTime { scope { account { id name } group { id name } site { id name } } }
|
||||
"""
|
||||
|
||||
|
||||
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 epoch_ms(v):
|
||||
"""Normalise a watermark to epoch milliseconds.
|
||||
|
||||
The alert-rule scheduler injects the watermark raw, in whichever format the
|
||||
operator picked for the rule (iso8601 | unix | unix_ms), so accept all three
|
||||
rather than forcing one and failing silently on the others.
|
||||
"""
|
||||
s = str(v or "").strip()
|
||||
if not s:
|
||||
return None
|
||||
if s.isdigit():
|
||||
n = int(s)
|
||||
return n * 1000 if n < 100000000000 else n # seconds vs milliseconds
|
||||
import datetime
|
||||
t = s.replace("Z", "+00:00")
|
||||
try:
|
||||
dt = datetime.datetime.fromisoformat(t)
|
||||
except ValueError:
|
||||
raise ValueError("cannot read timestamp %r: expected ISO8601 or an epoch" % s)
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=datetime.timezone.utc)
|
||||
return int(dt.timestamp() * 1000)
|
||||
|
||||
|
||||
def gql_string(v):
|
||||
"""Quote a value for inline use in the query (json.dumps escapes to GraphQL's rules)."""
|
||||
return json.dumps(str(v))
|
||||
|
||||
|
||||
def build_filters(inputs):
|
||||
out = []
|
||||
start = epoch_ms(inputs.get("created_after"))
|
||||
end = epoch_ms(inputs.get("created_until"))
|
||||
if start is not None or end is not None:
|
||||
parts = []
|
||||
if start is not None:
|
||||
parts.append("start: %d" % start)
|
||||
if end is not None:
|
||||
parts.append("end: %d" % end)
|
||||
out.append('{ fieldId: "createdAt" dateTimeRange: { %s } }' % " ".join(parts))
|
||||
updated = epoch_ms(inputs.get("updated_after"))
|
||||
if updated is not None:
|
||||
out.append('{ fieldId: "updatedAt" dateTimeRange: { start: %d } }' % updated)
|
||||
for field, key in (("status", "status"), ("severity", "severity"),
|
||||
("analystVerdict", "analyst_verdict"), ("id", "alert_id")):
|
||||
v = str(inputs.get(key) or "").strip()
|
||||
if v:
|
||||
out.append('{ fieldId: "%s" stringEqual: { value: %s } }' % (field, gql_string(v)))
|
||||
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"
|
||||
headers = {
|
||||
"Authorization": "ApiToken " + secrets.get("api_token", ""),
|
||||
"Accept": "application/json",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
view_type = str(inputs.get("view_type") or "ALL").strip().upper()
|
||||
if view_type not in VIEW_TYPES:
|
||||
raise ValueError("view_type must be one of %s" % ", ".join(VIEW_TYPES))
|
||||
|
||||
limit = int(inputs.get("limit") or 100)
|
||||
sort_by = "updatedAt" if str(inputs.get("updated_after") or "").strip() else "createdAt"
|
||||
filters = build_filters(inputs)
|
||||
filter_clause = "filters: [ %s ]" % " ".join(filters) if filters else ""
|
||||
|
||||
# Ascending, so a truncated run leaves the watermark on the oldest unseen
|
||||
# alert rather than skipping everything between it and the newest.
|
||||
nodes, cursor, page_info = [], str(inputs.get("cursor") or "").strip(), {}
|
||||
total = None
|
||||
while len(nodes) < limit:
|
||||
page = min(limit - len(nodes), 100)
|
||||
query = """query Alerts {
|
||||
alerts(
|
||||
first: %d
|
||||
viewType: %s
|
||||
after: %s
|
||||
sort: { by: "%s", order: ASC }
|
||||
%s
|
||||
) {
|
||||
totalCount
|
||||
edges { node { %s } }
|
||||
pageInfo { hasNextPage hasPreviousPage startCursor endCursor }
|
||||
}
|
||||
}""" % (page, view_type, gql_string(cursor) if cursor else "null", sort_by, filter_clause, NODE_FIELDS)
|
||||
|
||||
resp = request("POST", base + "/unifiedalerts/graphql", headers, {"query": query})
|
||||
# GraphQL answers 200 with an errors array; without this an unusable
|
||||
# query would look like an endless run of empty, successful fetches.
|
||||
if resp.get("errors"):
|
||||
print(json.dumps({"error": "GraphQL error", "detail": resp["errors"]}))
|
||||
sys.exit(1)
|
||||
|
||||
alerts = (resp.get("data") or {}).get("alerts") or {}
|
||||
if total is None:
|
||||
total = alerts.get("totalCount")
|
||||
edges = alerts.get("edges") or []
|
||||
for edge in edges:
|
||||
node = edge.get("node")
|
||||
if node:
|
||||
nodes.append(node)
|
||||
page_info = alerts.get("pageInfo") or {}
|
||||
cursor = page_info.get("endCursor")
|
||||
# An empty page ends the walk even when the console still claims a next
|
||||
# one: without this, a console answering hasNextPage on no edges would
|
||||
# keep the fetch spinning until the run is killed.
|
||||
if not edges or not page_info.get("hasNextPage") or not cursor:
|
||||
break
|
||||
|
||||
# Flattened to {data: [...]} so the alert rule reads a plain array of alerts
|
||||
# (results_path: data) and dedups on a top-level id, matching the REST
|
||||
# commands in this package. The edges/node envelope carries no id of its own.
|
||||
print(json.dumps({
|
||||
"data": nodes,
|
||||
"pagination": {
|
||||
"totalCount": total,
|
||||
"returned": len(nodes),
|
||||
"hasNextPage": bool(page_info.get("hasNextPage")),
|
||||
"endCursor": page_info.get("endCursor"),
|
||||
},
|
||||
}))
|
||||
|
||||
|
||||
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