feat(cortex-xdr): incident write-back and external alert push

Add three commands to the Cortex XDR integration (v1.2.0):
- update_incident: change status, severity, assignment and resolve
  comment on an incident (closes the SOAR ingest→action loop).
- insert_parsed_alerts: push external parsed alerts into Cortex XDR.
- insert_cef_alerts: push external CEF-format alerts.

Scripts follow the existing self-contained standard/advanced auth pattern.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

integrations/cortex-xdr/scripts/insert_cef_alerts.py

@@ -0,0 +1,54 @@
+import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
+
+
+def _client():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    base = s.get("url", "").rstrip("/") + "/public_api/v1"
+    key = s.get("api_key", "")
+    kid = str(s.get("api_key_id", ""))
+    headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
+    if str(s.get("auth_type") or "standard").lower() == "advanced":
+        nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
+        ts = str(int(time.time()) * 1000)
+        headers["x-xdr-nonce"] = nonce
+        headers["x-xdr-timestamp"] = ts
+        headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
+    else:
+        headers["Authorization"] = key
+    return base, headers
+
+
+def post(path, request_data):
+    base, headers = _client()
+    data = json.dumps({"request_data": request_data}).encode("utf-8")
+    req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    raw = inputs.get("alerts")
+    if isinstance(raw, list):
+        alerts = raw
+    else:
+        text = str(raw or "").strip()
+        try:
+            # A JSON array of CEF strings, when provided that way.
+            parsed = json.loads(text)
+            alerts = parsed if isinstance(parsed, list) else [text]
+        except Exception:
+            # Otherwise treat the input as newline-separated CEF lines.
+            alerts = [ln.strip() for ln in text.splitlines() if ln.strip()]
+    print(json.dumps(post("/alerts/insert_cef_alerts/", {"alerts": alerts})))
+
+
+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)

integrations/cortex-xdr/scripts/insert_parsed_alerts.py

@@ -0,0 +1,46 @@
+import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
+
+
+def _client():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    base = s.get("url", "").rstrip("/") + "/public_api/v1"
+    key = s.get("api_key", "")
+    kid = str(s.get("api_key_id", ""))
+    headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
+    if str(s.get("auth_type") or "standard").lower() == "advanced":
+        nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
+        ts = str(int(time.time()) * 1000)
+        headers["x-xdr-nonce"] = nonce
+        headers["x-xdr-timestamp"] = ts
+        headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
+    else:
+        headers["Authorization"] = key
+    return base, headers
+
+
+def post(path, request_data):
+    base, headers = _client()
+    data = json.dumps({"request_data": request_data}).encode("utf-8")
+    req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    raw = inputs.get("alerts")
+    alerts = raw if isinstance(raw, list) else json.loads(raw or "[]")
+    if not isinstance(alerts, list):
+        alerts = [alerts]
+    print(json.dumps(post("/alerts/insert_parsed_alerts/", {"alerts": alerts})))
+
+
+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)

integrations/cortex-xdr/scripts/update_incident.py

@@ -0,0 +1,57 @@
+import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
+
+
+def _client():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+    base = s.get("url", "").rstrip("/") + "/public_api/v1"
+    key = s.get("api_key", "")
+    kid = str(s.get("api_key_id", ""))
+    headers = {"x-xdr-auth-id": kid, "Content-Type": "application/json", "Accept": "application/json"}
+    if str(s.get("auth_type") or "standard").lower() == "advanced":
+        nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
+        ts = str(int(time.time()) * 1000)
+        headers["x-xdr-nonce"] = nonce
+        headers["x-xdr-timestamp"] = ts
+        headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
+    else:
+        headers["Authorization"] = key
+    return base, headers
+
+
+def post(path, request_data):
+    base, headers = _client()
+    data = json.dumps({"request_data": request_data}).encode("utf-8")
+    req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    update_data = {}
+    if inputs.get("status"):
+        update_data["status"] = inputs["status"]
+    if inputs.get("manual_severity"):
+        update_data["manual_severity"] = inputs["manual_severity"]
+    if inputs.get("assigned_user_mail"):
+        update_data["assigned_user_mail"] = inputs["assigned_user_mail"]
+    if inputs.get("assigned_user_pretty_name"):
+        update_data["assigned_user_pretty_name"] = inputs["assigned_user_pretty_name"]
+    if str(inputs.get("unassign")).lower() in ("true", "1", "yes"):
+        update_data["assigned_user_mail"] = None
+        update_data["assigned_user_pretty_name"] = None
+    if inputs.get("resolve_comment"):
+        update_data["resolve_comment"] = inputs["resolve_comment"]
+    rd = {"incident_id": inputs.get("incident_id"), "update_data": update_data}
+    print(json.dumps(post("/incidents/update_incident/", rd)))
+
+
+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)