f3nris
4cd1cda7cb
The SDL package covered the same unified (UAM) alerts from a second
integration: one console, two packages, two instances, two tokens. Its
commands now live here, so it can be dropped.
get_unified_alerts gains the SDL filters — classification, attack_surface,
os_type, asset_id, external_id, search_text, unassigned_only,
unmitigated_only and a detectedAt window. status, severity, analyst_verdict
and analytics_category take comma-separated lists (stringEqual on one value,
stringIn on several), and every time input also reads a relative window
('24 hours') on top of ISO8601 and epochs. The query is byte-identical to
1.9.0's for the inputs an existing alert rule sends, apart from enum values
now being upper-cased — a lowercase severity used to filter nothing.
Three capabilities the SDL package had and this one did not:
get_unified_alert_details (indicators, their MITRE mapping, observables,
related alerts), add_uam_alert_note and trigger_uam_alert_action (any console
action by ID). update_uam_alert_assignee completes the status/verdict pair,
on the same alertTriggerActions mutation.
analytics_category filters the fetch to one analytic category, e.g. only the
platform-rule alerts (analyticsCategory=Platform Rule, the console's own
filter). The SDL package showed this endpoint takes dotted fieldIds
(asset.osType, assignee.userId) and spells its sort argument differently, so
both spellings of each are tried before giving up: the category degrades to a
filter over the fetched alerts rather than failing the run, and
pagination.categoryFilter / .sortForm report which shape the console accepted.
update_uam_alert_status and _verdict now report a GraphQL error instead of
printing it as a successful update — a playbook that chained on a refused
update used to carry on as if it had gone through.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EU1Rk9N6vNtp3q7YvLpjcc
63 lines
2.6 KiB
Python
63 lines
2.6 KiB
Python
import json, os, sys, urllib.request, urllib.parse, urllib.error
|
|
|
|
# The single-alert view of the UAM model. get_unified_alerts reads the list
|
|
# shape; this adds what the list cannot carry per alert: the indicators behind
|
|
# the detection, their MITRE mapping, the observables and the sibling alerts.
|
|
QUERY = """query GetAlertDetails($id: ID!) {
|
|
alert(id: $id) {
|
|
id externalId name description severity status classification confidenceLevel
|
|
analystVerdict result detectedAt createdAt updatedAt firstSeenAt lastSeenAt
|
|
storylineId attackSurfaces noteExists
|
|
asset { id name osType osVersion agentVersion lastLoggedInUser }
|
|
assignee { userId fullName email }
|
|
detectionSource { vendor product engine }
|
|
process { cmdLine parentName username file { name path md5 sha1 sha256 size } }
|
|
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 }
|
|
}
|
|
}"""
|
|
|
|
|
|
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 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",
|
|
}
|
|
alert_id = str(inputs.get("alert_id") or "").strip()
|
|
if not alert_id:
|
|
raise ValueError("alert_id is required")
|
|
|
|
body = {"query": QUERY, "variables": {"id": alert_id}}
|
|
resp = request("POST", base + "/unifiedalerts/graphql", headers, body)
|
|
# GraphQL answers 200 with an errors array; reported as an error rather than
|
|
# printed as an empty alert, which would read as "this alert has no detail".
|
|
if resp.get("errors"):
|
|
print(json.dumps({"error": "GraphQL error", "detail": resp["errors"]}))
|
|
sys.exit(1)
|
|
print(json.dumps((resp.get("data") or {}).get("alert") or {}))
|
|
|
|
|
|
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)
|