f3nris
c67343b7cf
An instance can carry the account IDs of the client it serves, for clients sharing one multi-tenant console. count_agents, get_threats, get_alerts and get_unified_alerts use them when the block leaves its account filter empty; an account named on the block still wins. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TgjV26rz6VcP262dX4vBrD
460 lines
21 KiB
Python
460 lines
21 KiB
Python
import json, os, sys, time, 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")
|
|
|
|
# Walking the cursor is bounded: filtering client-side (see the attempt ladder) a
|
|
# page can be discarded almost whole, and without a cap a narrow filter on a busy
|
|
# console would page until the run is killed.
|
|
MAX_PAGES = 50
|
|
|
|
# The console narrows to platform-rule alerts with analyticsCategory=Platform
|
|
# Rule, but the fieldIds this endpoint takes are a mix of flat names (createdAt,
|
|
# externalId) and dotted paths (asset.osType, assignee.userId), so both spellings
|
|
# of the category are tried before falling back to filtering what was fetched.
|
|
CATEGORY_FIELD_IDS = ("analytics.category", "analyticsCategory")
|
|
|
|
# An alert carries its account twice: the scope it sits in now (realTime) and the
|
|
# one it was detected in (detectionTime). The first is what an operator means by
|
|
# "this account", so it is tried first, with the same degrade-or-filter-locally
|
|
# treatment as the category.
|
|
ACCOUNT_FIELD_IDS = ("realTime.scope.account.id", "detectionTime.scope.accountId")
|
|
|
|
# Same reason: the sort argument is spelled one way or the other depending on the
|
|
# console's schema version, and a wrong guess fails the whole query rather than
|
|
# degrading, so both are tried.
|
|
SORT_FORMS = ("sort", "sorts")
|
|
|
|
# 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 } } }
|
|
"""
|
|
|
|
# Only what the list nodes do not already carry, so merging a detail onto its
|
|
# node cannot overwrite the richer list fields with the detail's shorter ones.
|
|
DETAIL_FIELDS = """
|
|
indicators { uid type message severity eventTime
|
|
attacks { tactic { name uid } technique { name uid } }
|
|
observables { name value type typeName } }
|
|
observables { name value type typeName }
|
|
relatedAlerts { id name }
|
|
process { username file { size } }
|
|
"""
|
|
|
|
# Details are read in aliased batches: one alert per request would cost 100
|
|
# round trips on a default-limit fetch.
|
|
DETAIL_BATCH = 10
|
|
|
|
_UNITS = {"second": 1, "minute": 60, "hour": 3600, "day": 86400,
|
|
"week": 604800, "month": 2592000, "year": 31536000}
|
|
|
|
|
|
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 rel_seconds(text):
|
|
"""Read a relative window like '24 hours' or 'last 7 days' into seconds."""
|
|
num = unit = None
|
|
for t in str(text).lower().replace("last", "").split():
|
|
if t.isdigit():
|
|
num = int(t)
|
|
elif t.rstrip("s") in _UNITS:
|
|
unit = t.rstrip("s")
|
|
return num * _UNITS[unit] if (num is not None and unit) else None
|
|
|
|
|
|
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. A relative window
|
|
('24 hours') is accepted too, for a rule that reads a rolling window instead
|
|
of a watermark.
|
|
"""
|
|
s = str(v or "").strip()
|
|
if not s:
|
|
return None
|
|
secs = rel_seconds(s)
|
|
if secs is not None:
|
|
return int((time.time() - secs) * 1000)
|
|
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, an epoch or a window like '24 hours'" % 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 csv_list(v, upper=False):
|
|
"""Split a comma-separated input into values, dropping the empties."""
|
|
if v in (None, ""):
|
|
return []
|
|
items = v if isinstance(v, list) else str(v).split(",")
|
|
out = [str(x).strip() for x in items if str(x).strip()]
|
|
return [x.upper() for x in out] if upper else out
|
|
|
|
|
|
def truthy(v):
|
|
return str(v).strip().lower() in ("1", "true", "yes", "on")
|
|
|
|
|
|
def deep_merge(base, extra):
|
|
"""Fold extra into base without overwriting what base already says: the list
|
|
node is the authority on the fields both carry, the detail only fills in."""
|
|
for k, v in (extra or {}).items():
|
|
cur = base.get(k)
|
|
if isinstance(cur, dict) and isinstance(v, dict):
|
|
deep_merge(cur, v)
|
|
elif cur in (None, "", [], {}):
|
|
base[k] = v
|
|
return base
|
|
|
|
|
|
def fetch_details(base, headers, ids):
|
|
"""Read the per-alert detail for ids, as {id: detail}. Best effort: an alert
|
|
the console will not detail is left with its list fields rather than failing
|
|
the fetch that already succeeded."""
|
|
out, errors = {}, []
|
|
for start in range(0, len(ids), DETAIL_BATCH):
|
|
chunk = ids[start:start + DETAIL_BATCH]
|
|
aliases = {"d%d" % i: aid for i, aid in enumerate(chunk)}
|
|
query = "query AlertDetails {\n%s\n}" % "\n".join(
|
|
" %s: alert(id: %s) { %s }" % (alias, gql_string(aid), DETAIL_FIELDS)
|
|
for alias, aid in aliases.items())
|
|
resp = request("POST", base + "/unifiedalerts/graphql", headers, {"query": query})
|
|
for alias, value in ((resp.get("data") or {}) or {}).items():
|
|
if value and alias in aliases:
|
|
out[aliases[alias]] = value
|
|
if resp.get("errors"):
|
|
errors.extend(resp["errors"])
|
|
# A batch answers partially: one unreadable alert nulls its alias and
|
|
# leaves the rest. Retry only what is still missing, one at a time,
|
|
# so a single bad id does not cost the whole batch.
|
|
for aid in [a for a in chunk if a not in out]:
|
|
single = request("POST", base + "/unifiedalerts/graphql", headers,
|
|
{"query": "query AlertDetail { alert(id: %s) { %s } }"
|
|
% (gql_string(aid), DETAIL_FIELDS)})
|
|
value = (single.get("data") or {}).get("alert")
|
|
if value:
|
|
out[aid] = value
|
|
return out, errors
|
|
|
|
|
|
def norm(v):
|
|
"""Fold a category to letters/digits, so "Platform Rule", "platform rule" and
|
|
PLATFORM_RULE all compare equal (used only by the client-side fallback)."""
|
|
return "".join(ch for ch in str(v or "").lower() if ch.isalnum())
|
|
|
|
|
|
def category_matches(node, categories):
|
|
return norm(((node.get("analytics") or {}).get("category"))) in {norm(c) for c in categories}
|
|
|
|
|
|
def node_accounts(node):
|
|
"""Both account IDs the node carries, current scope first."""
|
|
out = []
|
|
real = (((node.get("realTime") or {}).get("scope") or {}).get("account") or {}).get("id")
|
|
detected = (((node.get("detectionTime") or {}).get("scope") or {}).get("accountId"))
|
|
for v in (real, detected):
|
|
if v not in (None, ""):
|
|
out.append(str(v).strip())
|
|
return out
|
|
|
|
|
|
def account_matches(node, accounts):
|
|
wanted = {str(a).strip() for a in accounts}
|
|
return any(a in wanted for a in node_accounts(node))
|
|
|
|
|
|
def string_filter(field, values):
|
|
"""Equality on one value, membership on several — an unfiltered field is the
|
|
caller's job to skip. stringEqual is kept for the single-value case because
|
|
that is what every console here is known to accept."""
|
|
if len(values) == 1:
|
|
return '{ fieldId: "%s" stringEqual: { value: %s } }' % (field, gql_string(values[0]))
|
|
return '{ fieldId: "%s" stringIn: { values: [%s] } }' % (
|
|
field, ", ".join(gql_string(v) for v in values))
|
|
|
|
|
|
def date_range_filter(field, start, end):
|
|
parts = []
|
|
if start is not None:
|
|
parts.append("start: %d" % start)
|
|
if end is not None:
|
|
parts.append("end: %d" % end)
|
|
return '{ fieldId: "%s" dateTimeRange: { %s } }' % (field, " ".join(parts))
|
|
|
|
|
|
def build_filters(inputs, category_field, account_field):
|
|
"""Assemble the filter list. category_field and account_field are the fieldIds
|
|
to filter on, or None to leave that one to the client-side pass."""
|
|
out = []
|
|
|
|
# ── Time windows ──────────────────────────────────────────────────────
|
|
for field, after_key, until_key in (("createdAt", "created_after", "created_until"),
|
|
("detectedAt", "detected_after", "detected_until")):
|
|
start = epoch_ms(inputs.get(after_key))
|
|
end = epoch_ms(inputs.get(until_key))
|
|
if start is not None or end is not None:
|
|
out.append(date_range_filter(field, start, end))
|
|
updated = epoch_ms(inputs.get("updated_after"))
|
|
if updated is not None:
|
|
out.append(date_range_filter("updatedAt", updated, None))
|
|
|
|
# ── Enum-valued fields: comma-separated, upper-cased ──────────────────
|
|
for field, key in (("status", "status"),
|
|
("severity", "severity"),
|
|
("analystVerdict", "analyst_verdict"),
|
|
("classification", "classification"),
|
|
("attackSurfaces", "attack_surface"),
|
|
("asset.osType", "os_type")):
|
|
values = csv_list(inputs.get(key), upper=True)
|
|
if values:
|
|
out.append(string_filter(field, values))
|
|
|
|
# ── Identifiers, taken as written ─────────────────────────────────────
|
|
for field, key in (("id", "alert_id"),
|
|
("asset.id", "asset_id"),
|
|
("externalId", "external_id")):
|
|
values = csv_list(inputs.get(key))
|
|
if values:
|
|
out.append(string_filter(field, values))
|
|
|
|
# ── The detecting analytic's category (e.g. Platform Rule) ────────────
|
|
if category_field:
|
|
values = csv_list(inputs.get("analytics_category"))
|
|
if values:
|
|
out.append(string_filter(category_field, values))
|
|
|
|
# ── Tenancy: the account the alert is scoped to ───────────────────────
|
|
if account_field:
|
|
values = csv_list(inputs.get("account_ids"))
|
|
if values:
|
|
out.append(string_filter(account_field, values))
|
|
|
|
# ── Free text and the two console toggles ─────────────────────────────
|
|
search = str(inputs.get("search_text") or "").strip()
|
|
if search:
|
|
out.append('{ fieldId: "name" match: { values: [%s] } }' % gql_string(search))
|
|
if truthy(inputs.get("unassigned_only")):
|
|
out.append('{ fieldId: "assignee.userId" isNegated: true stringEqual: { value: null } }')
|
|
if truthy(inputs.get("unmitigated_only")):
|
|
out.append('{ fieldId: "result" stringEqual: { value: "UNMITIGATED" } }')
|
|
return out
|
|
|
|
|
|
def sort_clause(form, by):
|
|
"""ASC either way, so a truncated run leaves the watermark on the oldest
|
|
unseen alert rather than skipping everything between it and the newest."""
|
|
if form == "sorts":
|
|
return 'sorts: [ { by: "%s", order: ASC } ]' % by
|
|
return 'sort: { by: "%s", order: ASC }' % by
|
|
|
|
|
|
def main():
|
|
secrets = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
|
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
|
# An instance bound to one account of a shared, multi-tenant console carries it
|
|
# in its config: it scopes the call unless the block names accounts itself.
|
|
if not str(inputs.get("account_ids") or "").strip() and str(secrets.get("account_ids") or "").strip():
|
|
inputs["account_ids"] = str(secrets["account_ids"])
|
|
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"
|
|
categories = csv_list(inputs.get("analytics_category"))
|
|
accounts = csv_list(inputs.get("account_ids"))
|
|
|
|
# Query shapes to try, in order, until the console accepts one. A filter that
|
|
# was not asked for has nothing to degrade, so it contributes no variants and
|
|
# the usual run settles on the first shape.
|
|
category_modes = list(CATEGORY_FIELD_IDS) + [None] if categories else [None]
|
|
account_modes = list(ACCOUNT_FIELD_IDS) + [None] if accounts else [None]
|
|
attempts = [(s, c, a) for s in SORT_FORMS for c in category_modes for a in account_modes]
|
|
attempt = 0
|
|
settled = False
|
|
|
|
nodes, cursor, page_info = [], str(inputs.get("cursor") or "").strip(), {}
|
|
total = None
|
|
pages = 0
|
|
stopped_mid_page = None
|
|
while len(nodes) < limit and pages < MAX_PAGES:
|
|
sort_form, category_field, account_field = attempts[attempt]
|
|
client_side_category = bool(categories) and category_field is None
|
|
client_side_accounts = bool(accounts) and account_field is None
|
|
# Server-side, every alert on a page is one we keep, so ask for exactly
|
|
# what is missing. Filtered client-side, most of a page is discarded, so
|
|
# read full pages rather than walking one alert at a time.
|
|
page = 100 if (client_side_category or client_side_accounts) else min(limit - len(nodes), 100)
|
|
page_cursor = cursor
|
|
filters = build_filters(inputs, category_field, account_field)
|
|
filter_clause = "filters: [ %s ]" % " ".join(filters) if filters else ""
|
|
query = """query Alerts {
|
|
alerts(
|
|
first: %d
|
|
viewType: %s
|
|
after: %s
|
|
%s
|
|
%s
|
|
) {
|
|
totalCount
|
|
edges { node { %s } }
|
|
pageInfo { hasNextPage hasPreviousPage startCursor endCursor }
|
|
}
|
|
}""" % (page, view_type, gql_string(cursor) if cursor else "null",
|
|
sort_clause(sort_form, 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"):
|
|
# A console that does not know a fieldId or the sort spelling answers
|
|
# with an errors array rather than ignoring it, so the next shape is
|
|
# tried on the same cursor. Once a page has come back the shape is
|
|
# settled and a later error is a real one.
|
|
if not settled and attempt + 1 < len(attempts):
|
|
attempt += 1
|
|
continue
|
|
print(json.dumps({"error": "GraphQL error", "detail": resp["errors"]}))
|
|
sys.exit(1)
|
|
|
|
settled = True
|
|
pages += 1
|
|
alerts = (resp.get("data") or {}).get("alerts") or {}
|
|
if total is None:
|
|
total = alerts.get("totalCount")
|
|
edges = alerts.get("edges") or []
|
|
for i, edge in enumerate(edges):
|
|
node = edge.get("node")
|
|
if not node:
|
|
continue
|
|
if client_side_category and not category_matches(node, categories):
|
|
continue
|
|
if client_side_accounts and not account_matches(node, accounts):
|
|
continue
|
|
nodes.append(node)
|
|
# Client-side, a page holds more alerts than the limit leaves room
|
|
# for. A cursor only addresses page boundaries, so stopping here
|
|
# reports the cursor this page was read with: the next run re-reads
|
|
# the page and the rule dedups on id, rather than the tail of the
|
|
# page being skipped for good.
|
|
if len(nodes) >= limit and i + 1 < len(edges):
|
|
stopped_mid_page = page_cursor
|
|
break
|
|
page_info = alerts.get("pageInfo") or {}
|
|
cursor = page_info.get("endCursor")
|
|
if stopped_mid_page is not None:
|
|
break
|
|
# 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
|
|
|
|
# Enrichment runs on the alerts actually returned, once the walk is over, so
|
|
# the alerts dropped by a client-side filter cost nothing.
|
|
details = {"requested": 0, "fetched": 0}
|
|
if truthy(inputs.get("with_details")) and nodes:
|
|
ids = [n.get("id") for n in nodes if n.get("id")]
|
|
details["requested"] = len(ids)
|
|
fetched, detail_errors = fetch_details(base, headers, ids)
|
|
details["fetched"] = len(fetched)
|
|
for node in nodes:
|
|
deep_merge(node, fetched.get(node.get("id")) or {})
|
|
# Alerts without their detail are still returned — an ingestion that got
|
|
# the alerts but not their indicators beats one that got nothing — so the
|
|
# shortfall is reported rather than raised.
|
|
if detail_errors and len(fetched) < len(ids):
|
|
details["error"] = detail_errors[0].get("message", str(detail_errors[0]))
|
|
|
|
sort_form, category_field, account_field = attempts[attempt]
|
|
# 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 is the console's count for the filters it applied, so
|
|
# with a client-side category it counts more than data[] holds.
|
|
"totalCount": total,
|
|
"returned": len(nodes),
|
|
"hasNextPage": True if stopped_mid_page is not None else bool(page_info.get("hasNextPage")),
|
|
"endCursor": (stopped_mid_page or None) if stopped_mid_page is not None else page_info.get("endCursor"),
|
|
"pagesRead": pages,
|
|
# Which query shape the console accepted — worth reading once per
|
|
# tenant: 'client' means the category was filtered after the fetch.
|
|
"categoryFilter": (None if not categories
|
|
else ("server:" + category_field if category_field else "client")),
|
|
"accountFilter": (None if not accounts
|
|
else ("server:" + account_field if account_field else "client")),
|
|
"sortForm": sort_form,
|
|
"details": details,
|
|
},
|
|
}))
|
|
|
|
|
|
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)
|