feat(cortex-xdr): new Cortex XDR integration — incident ingestion + OCSF mapper, IR actions, standard/advanced auth

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

integrations/cortex-xdr/scripts/action_status.py

@@ -0,0 +1,42 @@
+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", "{}"))
+    print(json.dumps(post("/actions/get_action_status/", {"group_action_id": inputs.get("action_id")})))
+
+
+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/allowlist_files.py

@@ -0,0 +1,51 @@
+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 csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    rd = {"hash_list": csv(inputs.get("hash_list"))}
+    if inputs.get("comment"):
+        rd["comment"] = inputs["comment"]
+    if inputs.get("incident_id"):
+        rd["incident_id"] = inputs["incident_id"]
+    print(json.dumps(post("/hash_exceptions/allowlist/", 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)

integrations/cortex-xdr/scripts/blocklist_files.py

@@ -0,0 +1,51 @@
+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 csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    rd = {"hash_list": csv(inputs.get("hash_list"))}
+    if inputs.get("comment"):
+        rd["comment"] = inputs["comment"]
+    if inputs.get("incident_id"):
+        rd["incident_id"] = inputs["incident_id"]
+    print(json.dumps(post("/hash_exceptions/blocklist/", 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)

integrations/cortex-xdr/scripts/get_alerts.py

@@ -0,0 +1,69 @@
+import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
+from datetime import datetime
+
+
+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 csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+def to_ms(v):
+    if v in (None, ""):
+        return None
+    s = str(v)
+    if s.isdigit():
+        return int(s)
+    try:
+        return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
+    except Exception:
+        return None
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    limit = int(inputs.get("limit") or 100)
+    filters = []
+    if inputs.get("severity"):
+        filters.append({"field": "severity", "operator": "in", "value": csv(inputs["severity"])})
+    created_ms = to_ms(inputs.get("created_after"))
+    if created_ms is not None:
+        filters.append({"field": "source_insert_ts", "operator": "gte", "value": created_ms})
+    rd = {"search_from": 0, "search_to": limit, "sort": {"field": "source_insert_ts", "keyword": "desc"}}
+    if filters:
+        rd["filters"] = filters
+    print(json.dumps(post("/alerts/get_alerts_by_filter_data/", 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)

integrations/cortex-xdr/scripts/get_distribution_versions.py

@@ -0,0 +1,41 @@
+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():
+    print(json.dumps(post("/distributions/get_versions/", {})))
+
+
+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/get_endpoints.py

@@ -0,0 +1,56 @@
+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 csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    limit = int(inputs.get("limit") or 30)
+    filters = []
+    if inputs.get("hostname"):
+        filters.append({"field": "hostname", "operator": "in", "value": csv(inputs["hostname"])})
+    if inputs.get("ip_list"):
+        filters.append({"field": "ip_list", "operator": "in", "value": csv(inputs["ip_list"])})
+    if inputs.get("status"):
+        filters.append({"field": "endpoint_status", "operator": "IN", "value": csv(inputs["status"])})
+    if inputs.get("platform"):
+        filters.append({"field": "platform", "operator": "in", "value": csv(inputs["platform"])})
+    print(json.dumps(post("/endpoints/get_endpoint/", {"search_from": 0, "search_to": limit, "filters": filters})))
+
+
+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/get_incident_extra_data.py

@@ -0,0 +1,43 @@
+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", "{}"))
+    rd = {"incident_id": inputs.get("incident_id"), "alerts_limit": int(inputs.get("alerts_limit") or 1000)}
+    print(json.dumps(post("/incidents/get_incident_extra_data/", 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)

integrations/cortex-xdr/scripts/get_incidents.py

@@ -0,0 +1,69 @@
+import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
+from datetime import datetime
+
+
+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 to_ms(v):
+    if v in (None, ""):
+        return None
+    s = str(v)
+    if s.isdigit():
+        return int(s)
+    try:
+        return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
+    except Exception:
+        return None
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    limit = int(inputs.get("limit") or 100)
+    filters = []
+    if inputs.get("status"):
+        filters.append({"field": "status", "operator": "eq", "value": inputs["status"]})
+    created_ms = to_ms(inputs.get("created_after"))
+    if created_ms is not None:
+        filters.append({"field": "creation_time", "operator": "gte", "value": created_ms})
+    request_data = {
+        "search_from": 0,
+        "search_to": limit,
+        "sort": {"field": "creation_time", "keyword": "desc"},
+    }
+    if filters:
+        request_data["filters"] = filters
+    print(json.dumps(post("/incidents/get_incidents/", request_data)))
+
+
+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/get_policy.py

@@ -0,0 +1,42 @@
+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", "{}"))
+    print(json.dumps(post("/endpoints/get_policy/", {"endpoint_id": inputs.get("endpoint_id")})))
+
+
+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/get_quarantine_status.py

@@ -0,0 +1,43 @@
+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", "{}"))
+    rd = {"files": [{"endpoint_id": inputs.get("endpoint_id"), "file_path": inputs.get("file_path"), "file_hash": inputs.get("file_hash")}]}
+    print(json.dumps(post("/quarantine/status/", 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)

integrations/cortex-xdr/scripts/get_script_execution_results.py

@@ -0,0 +1,42 @@
+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", "{}"))
+    print(json.dumps(post("/scripts/get_script_execution_results", {"action_id": inputs.get("action_id")})))
+
+
+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/get_scripts.py

@@ -0,0 +1,52 @@
+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 csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    filters = []
+    if inputs.get("name"):
+        filters.append({"field": "name", "operator": "in", "value": csv(inputs["name"])})
+    ws = inputs.get("windows_supported")
+    if ws not in (None, ""):
+        filters.append({"field": "windows_supported", "operator": "in", "value": [bool(ws)]})
+    print(json.dumps(post("/scripts/get_scripts/", {"filters": filters})))
+
+
+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/isolate_endpoint.py

@@ -0,0 +1,45 @@
+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", "{}"))
+    rd = {"endpoint_id": inputs.get("endpoint_id")}
+    if inputs.get("incident_id"):
+        rd["incident_id"] = inputs["incident_id"]
+    print(json.dumps(post("/endpoints/isolate", 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)

integrations/cortex-xdr/scripts/quarantine_files.py

@@ -0,0 +1,53 @@
+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 csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    rd = {
+        "filters": [{"field": "endpoint_id_list", "operator": "in", "value": csv(inputs.get("endpoint_id_list"))}],
+        "file_path": inputs.get("file_path"),
+        "file_hash": inputs.get("file_hash"),
+    }
+    if inputs.get("incident_id"):
+        rd["incident_id"] = inputs["incident_id"]
+    print(json.dumps(post("/endpoints/quarantine/", 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)

integrations/cortex-xdr/scripts/remove_allowlist_files.py

@@ -0,0 +1,51 @@
+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 csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    rd = {"hash_list": csv(inputs.get("hash_list"))}
+    if inputs.get("comment"):
+        rd["comment"] = inputs["comment"]
+    if inputs.get("incident_id"):
+        rd["incident_id"] = inputs["incident_id"]
+    print(json.dumps(post("/hash_exceptions/allowlist/remove/", 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)

integrations/cortex-xdr/scripts/remove_blocklist_files.py

@@ -0,0 +1,51 @@
+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 csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    rd = {"hash_list": csv(inputs.get("hash_list"))}
+    if inputs.get("comment"):
+        rd["comment"] = inputs["comment"]
+    if inputs.get("incident_id"):
+        rd["incident_id"] = inputs["incident_id"]
+    print(json.dumps(post("/hash_exceptions/blocklist/remove/", 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)

integrations/cortex-xdr/scripts/restore_file.py

@@ -0,0 +1,47 @@
+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", "{}"))
+    rd = {"file_hash": inputs.get("file_hash")}
+    if inputs.get("endpoint_id"):
+        rd["endpoint_id"] = inputs["endpoint_id"]
+    if inputs.get("incident_id"):
+        rd["incident_id"] = inputs["incident_id"]
+    print(json.dumps(post("/endpoints/restore/", 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)

integrations/cortex-xdr/scripts/run_script.py

@@ -0,0 +1,58 @@
+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 csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    try:
+        params = json.loads(inputs.get("parameters") or "{}")
+    except Exception:
+        params = {}
+    rd = {
+        "script_uid": inputs.get("script_uid"),
+        "timeout": int(inputs.get("timeout") or 600),
+        "filters": [{"field": "endpoint_id_list", "operator": "in", "value": csv(inputs.get("endpoint_ids"))}],
+        "parameters_values": params,
+    }
+    if inputs.get("incident_id"):
+        rd["incident_id"] = inputs["incident_id"]
+    print(json.dumps(post("/scripts/run_script/", 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)

integrations/cortex-xdr/scripts/scan_endpoints.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 csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    filters = []
+    if inputs.get("endpoint_id_list"):
+        filters.append({"field": "endpoint_id_list", "operator": "in", "value": csv(inputs["endpoint_id_list"])})
+    if inputs.get("hostname"):
+        filters.append({"field": "hostname", "operator": "in", "value": csv(inputs["hostname"])})
+    rd = {"filters": filters if filters else "all"}
+    if inputs.get("incident_id"):
+        rd["incident_id"] = inputs["incident_id"]
+    print(json.dumps(post("/endpoints/scan/", 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)

integrations/cortex-xdr/scripts/test_connection.py

@@ -0,0 +1,41 @@
+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():
+    print(json.dumps(post("/distributions/get_versions/", {})))
+
+
+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/unisolate_endpoint.py

@@ -0,0 +1,45 @@
+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", "{}"))
+    rd = {"endpoint_id": inputs.get("endpoint_id")}
+    if inputs.get("incident_id"):
+        rd["incident_id"] = inputs["incident_id"]
+    print(json.dumps(post("/endpoints/unisolate", 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)