From 76812ef128ab6c687cec5867d222de2d654c2912 Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Fri, 26 Jun 2026 15:41:59 +0200 Subject: [PATCH] =?UTF-8?q?feat(cortex-xdr):=20full=20command=20coverage?= =?UTF-8?q?=20(50=20commands)=20=E2=80=94=20endpoints,=20scripts,=20exclus?= =?UTF-8?q?ions,=20tags,=20audits,=20RBAC,=20distributions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 --- integrations/cortex-xdr/manifest.yaml | 279 +++++++++++++++++- integrations/cortex-xdr/scripts/abort_scan.py | 69 +++++ .../cortex-xdr/scripts/add_exclusion.py | 64 ++++ .../cortex-xdr/scripts/add_tag_endpoint.py | 65 ++++ .../scripts/audit_management_logs.py | 74 +++++ .../cortex-xdr/scripts/create_distribution.py | 64 ++++ .../cortex-xdr/scripts/delete_endpoints.py | 64 ++++ .../cortex-xdr/scripts/delete_exclusion.py | 63 ++++ .../scripts/get_audit_agent_reports.py | 74 +++++ .../scripts/get_device_control_violations.py | 71 +++++ .../scripts/get_distribution_status.py | 63 ++++ .../scripts/get_distribution_url.py | 63 ++++ .../cortex-xdr/scripts/get_exclusion.py | 66 +++++ .../cortex-xdr/scripts/get_original_alerts.py | 63 ++++ .../cortex-xdr/scripts/get_risk_score.py | 63 ++++ .../cortex-xdr/scripts/get_script_code.py | 63 ++++ .../scripts/get_script_execution_status.py | 63 ++++ .../cortex-xdr/scripts/get_script_metadata.py | 63 ++++ .../cortex-xdr/scripts/list_risky_hosts.py | 62 ++++ .../cortex-xdr/scripts/list_risky_users.py | 62 ++++ integrations/cortex-xdr/scripts/list_roles.py | 63 ++++ .../cortex-xdr/scripts/list_user_groups.py | 63 ++++ integrations/cortex-xdr/scripts/list_users.py | 62 ++++ .../cortex-xdr/scripts/remove_tag_endpoint.py | 65 ++++ .../cortex-xdr/scripts/remove_user_role.py | 63 ++++ .../cortex-xdr/scripts/retrieve_file.py | 74 +++++ .../scripts/retrieve_file_details.py | 63 ++++ .../scripts/run_snippet_code_script.py | 66 +++++ .../cortex-xdr/scripts/set_endpoint_alias.py | 64 ++++ .../cortex-xdr/scripts/set_user_role.py | 63 ++++ 30 files changed, 2161 insertions(+), 3 deletions(-) create mode 100644 integrations/cortex-xdr/scripts/abort_scan.py create mode 100644 integrations/cortex-xdr/scripts/add_exclusion.py create mode 100644 integrations/cortex-xdr/scripts/add_tag_endpoint.py create mode 100644 integrations/cortex-xdr/scripts/audit_management_logs.py create mode 100644 integrations/cortex-xdr/scripts/create_distribution.py create mode 100644 integrations/cortex-xdr/scripts/delete_endpoints.py create mode 100644 integrations/cortex-xdr/scripts/delete_exclusion.py create mode 100644 integrations/cortex-xdr/scripts/get_audit_agent_reports.py create mode 100644 integrations/cortex-xdr/scripts/get_device_control_violations.py create mode 100644 integrations/cortex-xdr/scripts/get_distribution_status.py create mode 100644 integrations/cortex-xdr/scripts/get_distribution_url.py create mode 100644 integrations/cortex-xdr/scripts/get_exclusion.py create mode 100644 integrations/cortex-xdr/scripts/get_original_alerts.py create mode 100644 integrations/cortex-xdr/scripts/get_risk_score.py create mode 100644 integrations/cortex-xdr/scripts/get_script_code.py create mode 100644 integrations/cortex-xdr/scripts/get_script_execution_status.py create mode 100644 integrations/cortex-xdr/scripts/get_script_metadata.py create mode 100644 integrations/cortex-xdr/scripts/list_risky_hosts.py create mode 100644 integrations/cortex-xdr/scripts/list_risky_users.py create mode 100644 integrations/cortex-xdr/scripts/list_roles.py create mode 100644 integrations/cortex-xdr/scripts/list_user_groups.py create mode 100644 integrations/cortex-xdr/scripts/list_users.py create mode 100644 integrations/cortex-xdr/scripts/remove_tag_endpoint.py create mode 100644 integrations/cortex-xdr/scripts/remove_user_role.py create mode 100644 integrations/cortex-xdr/scripts/retrieve_file.py create mode 100644 integrations/cortex-xdr/scripts/retrieve_file_details.py create mode 100644 integrations/cortex-xdr/scripts/run_snippet_code_script.py create mode 100644 integrations/cortex-xdr/scripts/set_endpoint_alias.py create mode 100644 integrations/cortex-xdr/scripts/set_user_role.py diff --git a/integrations/cortex-xdr/manifest.yaml b/integrations/cortex-xdr/manifest.yaml index f4fa70b..d92b7e5 100644 --- a/integrations/cortex-xdr/manifest.yaml +++ b/integrations/cortex-xdr/manifest.yaml @@ -1,8 +1,8 @@ id: cortex_xdr name: Cortex XDR -version: 1.0.0 -description: "Palo Alto Cortex XDR (public API v1) — incident ingestion, endpoint isolation, RTR scripts, hash block/allow lists, file quarantine/restore, scans and alert retrieval." -changelog: "1.0.0 — Initial release: incident ingestion (get_incidents) with OCSF mapper, endpoints, isolate/unisolate, scan, hash blocklist/allowlist, quarantine/restore, run script + results, alerts retrieval, distributions and action status. Standard or Advanced API authentication." +version: 1.1.0 +description: "Palo Alto Cortex XDR (public API v1) — incident ingestion, endpoint isolation/scan/delete/tagging, RTR scripts, hash block/allow lists, file quarantine/restore/retrieval, alert exclusions, device-control violations, audits, distributions and RBAC/risk." +changelog: "1.1.0 — Full command coverage: added delete/alias/tag endpoints, abort scan, original alerts, script metadata/code/snippet/exec-status, file retrieval (+details), alert exclusions, device-control violations, audits, distribution url/status/create and RBAC (users, roles, groups, risk score, risky users/hosts). 1.0.0 — Initial release: incident ingestion (get_incidents) with OCSF mapper, endpoints, isolate/unisolate, scan, hash blocklist/allowlist, quarantine/restore, run script + results, alerts retrieval, distributions and action status. Standard or Advanced API authentication." category: endpoint # Per-instance configuration. The base URL is the tenant API root, e.g. @@ -260,6 +260,279 @@ commands: required: [] outputs_schema: { properties: {} } + - id: delete_endpoints + name: cortex-xdr-delete-endpoints + description: "Delete endpoints (agents) by ID." + inputs_schema: + properties: + endpoint_ids: { type: string, description: "Comma-separated endpoint IDs to delete" } + required: [endpoint_ids] + outputs_schema: { properties: {} } + - id: set_endpoint_alias + name: cortex-xdr-set-endpoint-alias + description: "Set the alias name on endpoints." + inputs_schema: + properties: + endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" } + alias: { type: string, description: "New alias name (empty to clear)" } + required: [endpoint_ids] + outputs_schema: { properties: {} } + - id: abort_scan + name: cortex-xdr-abort-scan + description: "Abort a running malware scan on endpoints matching the filters (or all)." + inputs_schema: + properties: + endpoint_id_list: { type: string, description: "Comma-separated endpoint IDs (omit to abort all)" } + incident_id: { type: string, description: "Optional incident ID" } + required: [] + outputs_schema: { properties: {} } + - id: get_original_alerts + name: cortex-xdr-get-original-alerts + description: "Get the original raw alert events by alert ID." + risk: read + inputs_schema: + properties: + alert_id_list: { type: string, description: "Comma-separated alert IDs" } + required: [alert_id_list] + outputs_schema: { properties: {} } + - id: get_script_metadata + name: cortex-xdr-get-script-metadata + description: "Get metadata for a script by UID." + risk: read + inputs_schema: + properties: + script_uid: { type: string, description: "Script UID" } + required: [script_uid] + outputs_schema: { properties: {} } + - id: get_script_code + name: cortex-xdr-get-script-code + description: "Get the source code of a script by UID." + risk: read + inputs_schema: + properties: + script_uid: { type: string, description: "Script UID" } + required: [script_uid] + outputs_schema: { properties: {} } + - id: run_snippet_code_script + name: cortex-xdr-run-snippet-code-script + description: "Run an arbitrary code snippet on endpoints." + inputs_schema: + properties: + snippet_code: { type: string, description: "Code snippet to run" } + endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" } + incident_id: { type: string, description: "Optional incident ID" } + required: [snippet_code, endpoint_ids] + outputs_schema: { properties: {} } + - id: get_script_execution_status + name: cortex-xdr-get-script-execution-status + description: "Get the status of a script execution by action ID." + risk: read + inputs_schema: + properties: + action_id: { type: string, description: "Action ID" } + required: [action_id] + outputs_schema: { properties: {} } + - id: retrieve_file + name: cortex-xdr-retrieve-file + description: "Retrieve files from endpoints by OS-specific path lists." + inputs_schema: + properties: + endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" } + windows: { type: string, description: "Comma-separated Windows file paths" } + linux: { type: string, description: "Comma-separated Linux file paths" } + macos: { type: string, description: "Comma-separated macOS file paths" } + incident_id: { type: string, description: "Optional incident ID" } + required: [endpoint_ids] + outputs_schema: { properties: {} } + - id: retrieve_file_details + name: cortex-xdr-retrieve-file-details + description: "Get the download details of a file-retrieval action." + risk: read + inputs_schema: + properties: + action_id: { type: string, description: "Group action ID from retrieve_file" } + required: [action_id] + outputs_schema: { properties: {} } + - id: add_exclusion + name: cortex-xdr-add-exclusion + description: "Add an alert exclusion." + inputs_schema: + properties: + name: { type: string, description: "Exclusion name" } + indicator: { type: string, description: "JSON indicator object for the exclusion filter" } + status: { type: string, description: "ENABLED or DISABLED (default ENABLED)" } + required: [name, indicator] + outputs_schema: { properties: {} } + - id: delete_exclusion + name: cortex-xdr-delete-exclusion + description: "Delete an alert exclusion by ID." + inputs_schema: + properties: + alert_exclusion_id: { type: number, description: "Exclusion ID" } + required: [alert_exclusion_id] + outputs_schema: { properties: {} } + - id: get_exclusion + name: cortex-xdr-get-exclusion + description: "List alert exclusions." + risk: read + inputs_schema: + properties: + tenant_id: { type: string, description: "Optional tenant ID" } + required: [] + outputs_schema: { properties: {} } + - id: add_tag_endpoint + name: cortex-xdr-add-tag-endpoint + description: "Add a tag to endpoints." + inputs_schema: + properties: + endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" } + tag: { type: string, description: "Tag to add" } + required: [endpoint_ids, tag] + outputs_schema: { properties: {} } + - id: remove_tag_endpoint + name: cortex-xdr-remove-tag-endpoint + description: "Remove a tag from endpoints." + inputs_schema: + properties: + endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" } + tag: { type: string, description: "Tag to remove" } + required: [endpoint_ids, tag] + outputs_schema: { properties: {} } + - id: get_device_control_violations + name: cortex-xdr-get-device-control-violations + description: "List device control (USB) violations." + risk: read + inputs_schema: + properties: + endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" } + type: { type: string, description: "Comma-separated violation types" } + timestamp_gte: { type: string, description: "Lower bound timestamp (ISO8601 or epoch ms)" } + required: [] + outputs_schema: { properties: {} } + - id: list_users + name: cortex-xdr-list-users + description: "List Cortex XDR users." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } + - id: get_risk_score + name: cortex-xdr-get-risk-score + description: "Get the risk score of a user or host by ID." + risk: read + inputs_schema: + properties: + id: { type: string, description: "User or host identifier" } + required: [id] + outputs_schema: { properties: {} } + - id: list_risky_users + name: cortex-xdr-list-risky-users + description: "List risky users." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } + - id: list_risky_hosts + name: cortex-xdr-list-risky-hosts + description: "List risky hosts." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } + - id: list_user_groups + name: cortex-xdr-list-user-groups + description: "Get user groups by name." + risk: read + inputs_schema: + properties: + group_names: { type: string, description: "Comma-separated group names" } + required: [] + outputs_schema: { properties: {} } + - id: list_roles + name: cortex-xdr-list-roles + description: "Get roles by name." + risk: read + inputs_schema: + properties: + role_names: { type: string, description: "Comma-separated role names" } + required: [] + outputs_schema: { properties: {} } + - id: set_user_role + name: cortex-xdr-set-user-role + description: "Assign a role to users." + inputs_schema: + properties: + user_emails: { type: string, description: "Comma-separated user emails" } + role_name: { type: string, description: "Role name to assign" } + required: [user_emails, role_name] + outputs_schema: { properties: {} } + - id: remove_user_role + name: cortex-xdr-remove-user-role + description: "Remove the role assignment from users." + inputs_schema: + properties: + user_emails: { type: string, description: "Comma-separated user emails" } + required: [user_emails] + outputs_schema: { properties: {} } + - id: audit_management_logs + name: cortex-xdr-audit-management-logs + description: "Retrieve management audit logs." + risk: read + inputs_schema: + properties: + email: { type: string, description: "Comma-separated emails" } + type: { type: string, description: "Comma-separated log types" } + timestamp_gte: { type: string, description: "Lower bound timestamp (ISO8601 or epoch ms)" } + limit: { type: number, description: "Max records (default 100)" } + required: [] + outputs_schema: { properties: {} } + - id: get_audit_agent_reports + name: cortex-xdr-get-audit-agent-reports + description: "Retrieve agent audit reports." + risk: read + inputs_schema: + properties: + endpoint_ids: { type: string, description: "Comma-separated endpoint IDs" } + type: { type: string, description: "Comma-separated report types" } + timestamp_gte: { type: string, description: "Lower bound timestamp (ISO8601 or epoch ms)" } + limit: { type: number, description: "Max records (default 100)" } + required: [] + outputs_schema: { properties: {} } + - id: get_distribution_url + name: cortex-xdr-get-distribution-url + description: "Get the download URL of an agent installer distribution." + risk: read + inputs_schema: + properties: + distribution_id: { type: string, description: "Distribution ID" } + package_type: { type: string, description: "Package type (sh, rpm, deb, pkg, x86, x64, ...)" } + required: [distribution_id, package_type] + outputs_schema: { properties: {} } + - id: get_distribution_status + name: cortex-xdr-get-distribution-status + description: "Get the build status of an agent installer distribution." + risk: read + inputs_schema: + properties: + distribution_id: { type: string, description: "Distribution ID" } + required: [distribution_id] + outputs_schema: { properties: {} } + - id: create_distribution + name: cortex-xdr-create-distribution + description: "Create an agent installer distribution." + inputs_schema: + properties: + name: { type: string, description: "Distribution name" } + platform: { type: string, description: "Platform (windows, linux, macos, android)" } + package_type: { type: string, description: "standalone or upgrade (default standalone)" } + agent_version: { type: string, description: "Agent version" } + description: { type: string, description: "Optional description" } + required: [name, platform] + outputs_schema: { properties: {} } + - id: test_connection name: cortex-xdr-test-connection description: "Verify connectivity and credentials (used by the Test button)." diff --git a/integrations/cortex-xdr/scripts/abort_scan.py b/integrations/cortex-xdr/scripts/abort_scan.py new file mode 100644 index 0000000..c9784c5 --- /dev/null +++ b/integrations/cortex-xdr/scripts/abort_scan.py @@ -0,0 +1,69 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + filters = [] + if inputs.get("endpoint_id_list"): + filters.append({"field": "endpoint_id_list", "operator": "in", "value": csv(inputs["endpoint_id_list"])}) + rd = {"filters": filters if filters else "all"} + if inputs.get("incident_id"): + rd["incident_id"] = inputs["incident_id"] + print(json.dumps(post("/endpoints/abort_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) diff --git a/integrations/cortex-xdr/scripts/add_exclusion.py b/integrations/cortex-xdr/scripts/add_exclusion.py new file mode 100644 index 0000000..7e7faab --- /dev/null +++ b/integrations/cortex-xdr/scripts/add_exclusion.py @@ -0,0 +1,64 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + rd = {"indicator": json.loads(inputs.get("indicator") or "{}"), "name": inputs.get("name"), "status": inputs.get("status") or "ENABLED"} + print(json.dumps(post("/alerts_exclusion/add/", 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) diff --git a/integrations/cortex-xdr/scripts/add_tag_endpoint.py b/integrations/cortex-xdr/scripts/add_tag_endpoint.py new file mode 100644 index 0000000..7956fc8 --- /dev/null +++ b/integrations/cortex-xdr/scripts/add_tag_endpoint.py @@ -0,0 +1,65 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + eids = csv(inputs.get("endpoint_ids")) + body = {"context": {"lcaas_id": eids}, "request_data": {"filters": [{"field": "endpoint_id_list", "operator": "in", "value": eids}], "tag": inputs.get("tag")}} + print(json.dumps(post_raw("/tags/agents/assign/", body))) + + +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) diff --git a/integrations/cortex-xdr/scripts/audit_management_logs.py b/integrations/cortex-xdr/scripts/audit_management_logs.py new file mode 100644 index 0000000..7d36a56 --- /dev/null +++ b/integrations/cortex-xdr/scripts/audit_management_logs.py @@ -0,0 +1,74 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + filters = [] + if inputs.get("email"): + filters.append({"field": "email", "operator": "in", "value": csv(inputs["email"])}) + if inputs.get("type"): + filters.append({"field": "type", "operator": "in", "value": csv(inputs["type"])}) + g = to_ms(inputs.get("timestamp_gte")) + if g is not None: + filters.append({"field": "timestamp", "operator": "gte", "value": g}) + rd = {"search_from": 0, "search_to": int(inputs.get("limit") or 100)} + if filters: + rd["filters"] = filters + print(json.dumps(post("/audits/management_logs/", 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) diff --git a/integrations/cortex-xdr/scripts/create_distribution.py b/integrations/cortex-xdr/scripts/create_distribution.py new file mode 100644 index 0000000..472414c --- /dev/null +++ b/integrations/cortex-xdr/scripts/create_distribution.py @@ -0,0 +1,64 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + rd = {"name": inputs.get("name"), "platform": inputs.get("platform"), "package_type": inputs.get("package_type") or "standalone", "agent_version": inputs.get("agent_version"), "description": inputs.get("description") or ""} + print(json.dumps(post("/distributions/create/", 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) diff --git a/integrations/cortex-xdr/scripts/delete_endpoints.py b/integrations/cortex-xdr/scripts/delete_endpoints.py new file mode 100644 index 0000000..d89ae96 --- /dev/null +++ b/integrations/cortex-xdr/scripts/delete_endpoints.py @@ -0,0 +1,64 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + rd = {"filters": [{"field": "endpoint_id_list", "operator": "in", "value": csv(inputs.get("endpoint_ids"))}]} + print(json.dumps(post("/endpoints/delete/", 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) diff --git a/integrations/cortex-xdr/scripts/delete_exclusion.py b/integrations/cortex-xdr/scripts/delete_exclusion.py new file mode 100644 index 0000000..a0736d1 --- /dev/null +++ b/integrations/cortex-xdr/scripts/delete_exclusion.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/alerts_exclusion/delete/", {"alert_exclusion_id": int(inputs.get("alert_exclusion_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) diff --git a/integrations/cortex-xdr/scripts/get_audit_agent_reports.py b/integrations/cortex-xdr/scripts/get_audit_agent_reports.py new file mode 100644 index 0000000..3c50e60 --- /dev/null +++ b/integrations/cortex-xdr/scripts/get_audit_agent_reports.py @@ -0,0 +1,74 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + filters = [] + if inputs.get("endpoint_ids"): + filters.append({"field": "endpoint_id", "operator": "in", "value": csv(inputs["endpoint_ids"])}) + if inputs.get("type"): + filters.append({"field": "type", "operator": "in", "value": csv(inputs["type"])}) + g = to_ms(inputs.get("timestamp_gte")) + if g is not None: + filters.append({"field": "timestamp", "operator": "gte", "value": g}) + rd = {"search_from": 0, "search_to": int(inputs.get("limit") or 100)} + if filters: + rd["filters"] = filters + print(json.dumps(post("/audits/agents_reports/", 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) diff --git a/integrations/cortex-xdr/scripts/get_device_control_violations.py b/integrations/cortex-xdr/scripts/get_device_control_violations.py new file mode 100644 index 0000000..d530229 --- /dev/null +++ b/integrations/cortex-xdr/scripts/get_device_control_violations.py @@ -0,0 +1,71 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + filters = [] + if inputs.get("endpoint_ids"): + filters.append({"field": "endpoint_id_list", "operator": "in", "value": csv(inputs["endpoint_ids"])}) + if inputs.get("type"): + filters.append({"field": "type", "operator": "in", "value": csv(inputs["type"])}) + g = to_ms(inputs.get("timestamp_gte")) + if g is not None: + filters.append({"field": "timestamp", "operator": "gte", "value": g}) + print(json.dumps(post("/device_control/get_violations/", {"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) diff --git a/integrations/cortex-xdr/scripts/get_distribution_status.py b/integrations/cortex-xdr/scripts/get_distribution_status.py new file mode 100644 index 0000000..817b73c --- /dev/null +++ b/integrations/cortex-xdr/scripts/get_distribution_status.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/distributions/get_status/", {"distribution_id": inputs.get("distribution_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) diff --git a/integrations/cortex-xdr/scripts/get_distribution_url.py b/integrations/cortex-xdr/scripts/get_distribution_url.py new file mode 100644 index 0000000..ff720e0 --- /dev/null +++ b/integrations/cortex-xdr/scripts/get_distribution_url.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/distributions/get_dist_url/", {"distribution_id": inputs.get("distribution_id"), "package_type": inputs.get("package_type")}))) + + +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) diff --git a/integrations/cortex-xdr/scripts/get_exclusion.py b/integrations/cortex-xdr/scripts/get_exclusion.py new file mode 100644 index 0000000..bb10c74 --- /dev/null +++ b/integrations/cortex-xdr/scripts/get_exclusion.py @@ -0,0 +1,66 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + rd = {} + if inputs.get("tenant_id"): + rd["tenant_id"] = inputs["tenant_id"] + print(json.dumps(post("/alerts_exclusion/", 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) diff --git a/integrations/cortex-xdr/scripts/get_original_alerts.py b/integrations/cortex-xdr/scripts/get_original_alerts.py new file mode 100644 index 0000000..9d002ff --- /dev/null +++ b/integrations/cortex-xdr/scripts/get_original_alerts.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/alerts/get_original_alerts/", {"alert_id_list": csv(inputs.get("alert_id_list"))}))) + + +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) diff --git a/integrations/cortex-xdr/scripts/get_risk_score.py b/integrations/cortex-xdr/scripts/get_risk_score.py new file mode 100644 index 0000000..a854296 --- /dev/null +++ b/integrations/cortex-xdr/scripts/get_risk_score.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/get_risk_score/", {"id": inputs.get("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) diff --git a/integrations/cortex-xdr/scripts/get_script_code.py b/integrations/cortex-xdr/scripts/get_script_code.py new file mode 100644 index 0000000..271ffaa --- /dev/null +++ b/integrations/cortex-xdr/scripts/get_script_code.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/scripts/get_script_code/", {"script_uid": inputs.get("script_uid")}))) + + +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) diff --git a/integrations/cortex-xdr/scripts/get_script_execution_status.py b/integrations/cortex-xdr/scripts/get_script_execution_status.py new file mode 100644 index 0000000..440d0d7 --- /dev/null +++ b/integrations/cortex-xdr/scripts/get_script_execution_status.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/scripts/get_script_execution_status/", {"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) diff --git a/integrations/cortex-xdr/scripts/get_script_metadata.py b/integrations/cortex-xdr/scripts/get_script_metadata.py new file mode 100644 index 0000000..ff14f07 --- /dev/null +++ b/integrations/cortex-xdr/scripts/get_script_metadata.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/scripts/get_script_metadata/", {"script_uid": inputs.get("script_uid")}))) + + +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) diff --git a/integrations/cortex-xdr/scripts/list_risky_hosts.py b/integrations/cortex-xdr/scripts/list_risky_hosts.py new file mode 100644 index 0000000..1d7e924 --- /dev/null +++ b/integrations/cortex-xdr/scripts/list_risky_hosts.py @@ -0,0 +1,62 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000) + except Exception: + return None + + +def main(): + print(json.dumps(post("/get_risky_hosts/"))) + + +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) diff --git a/integrations/cortex-xdr/scripts/list_risky_users.py b/integrations/cortex-xdr/scripts/list_risky_users.py new file mode 100644 index 0000000..773f792 --- /dev/null +++ b/integrations/cortex-xdr/scripts/list_risky_users.py @@ -0,0 +1,62 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000) + except Exception: + return None + + +def main(): + print(json.dumps(post("/get_risky_users/"))) + + +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) diff --git a/integrations/cortex-xdr/scripts/list_roles.py b/integrations/cortex-xdr/scripts/list_roles.py new file mode 100644 index 0000000..0dd92f9 --- /dev/null +++ b/integrations/cortex-xdr/scripts/list_roles.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/rbac/get_roles/", {"role_names": csv(inputs.get("role_names"))}))) + + +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) diff --git a/integrations/cortex-xdr/scripts/list_user_groups.py b/integrations/cortex-xdr/scripts/list_user_groups.py new file mode 100644 index 0000000..3f3961a --- /dev/null +++ b/integrations/cortex-xdr/scripts/list_user_groups.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/rbac/get_user_group/", {"group_names": csv(inputs.get("group_names"))}))) + + +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) diff --git a/integrations/cortex-xdr/scripts/list_users.py b/integrations/cortex-xdr/scripts/list_users.py new file mode 100644 index 0000000..43cd549 --- /dev/null +++ b/integrations/cortex-xdr/scripts/list_users.py @@ -0,0 +1,62 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000) + except Exception: + return None + + +def main(): + print(json.dumps(post("/rbac/get_users/", {}))) + + +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) diff --git a/integrations/cortex-xdr/scripts/remove_tag_endpoint.py b/integrations/cortex-xdr/scripts/remove_tag_endpoint.py new file mode 100644 index 0000000..6995ee9 --- /dev/null +++ b/integrations/cortex-xdr/scripts/remove_tag_endpoint.py @@ -0,0 +1,65 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + eids = csv(inputs.get("endpoint_ids")) + body = {"context": {"lcaas_id": eids}, "request_data": {"filters": [{"field": "endpoint_id_list", "operator": "in", "value": eids}], "tag": inputs.get("tag")}} + print(json.dumps(post_raw("/tags/agents/remove/", body))) + + +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) diff --git a/integrations/cortex-xdr/scripts/remove_user_role.py b/integrations/cortex-xdr/scripts/remove_user_role.py new file mode 100644 index 0000000..5420513 --- /dev/null +++ b/integrations/cortex-xdr/scripts/remove_user_role.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/rbac/set_user_role/", {"user_emails": csv(inputs.get("user_emails"))}))) + + +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) diff --git a/integrations/cortex-xdr/scripts/retrieve_file.py b/integrations/cortex-xdr/scripts/retrieve_file.py new file mode 100644 index 0000000..d9265e2 --- /dev/null +++ b/integrations/cortex-xdr/scripts/retrieve_file.py @@ -0,0 +1,74 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + eids = csv(inputs.get("endpoint_ids")) + files = {} + if inputs.get("windows"): + files["windows"] = csv(inputs["windows"]) + if inputs.get("linux"): + files["linux"] = csv(inputs["linux"]) + if inputs.get("macos"): + files["macos"] = csv(inputs["macos"]) + rd = {"filters": [{"field": "endpoint_id_list", "operator": "in", "value": eids}], "files": files} + if inputs.get("incident_id"): + rd["incident_id"] = inputs["incident_id"] + print(json.dumps(post("/endpoints/file_retrieval/", 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) diff --git a/integrations/cortex-xdr/scripts/retrieve_file_details.py b/integrations/cortex-xdr/scripts/retrieve_file_details.py new file mode 100644 index 0000000..c2c66e3 --- /dev/null +++ b/integrations/cortex-xdr/scripts/retrieve_file_details.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/actions/file_retrieval_details/", {"group_action_id": int(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) diff --git a/integrations/cortex-xdr/scripts/run_snippet_code_script.py b/integrations/cortex-xdr/scripts/run_snippet_code_script.py new file mode 100644 index 0000000..e3c4ed8 --- /dev/null +++ b/integrations/cortex-xdr/scripts/run_snippet_code_script.py @@ -0,0 +1,66 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + rd = {"filters": [{"field": "endpoint_id_list", "operator": "in", "value": csv(inputs.get("endpoint_ids"))}], "snippet_code": inputs.get("snippet_code")} + if inputs.get("incident_id"): + rd["incident_id"] = inputs["incident_id"] + print(json.dumps(post("/scripts/run_snippet_code_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) diff --git a/integrations/cortex-xdr/scripts/set_endpoint_alias.py b/integrations/cortex-xdr/scripts/set_endpoint_alias.py new file mode 100644 index 0000000..a839571 --- /dev/null +++ b/integrations/cortex-xdr/scripts/set_endpoint_alias.py @@ -0,0 +1,64 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + rd = {"filters": [{"field": "endpoint_id_list", "operator": "in", "value": csv(inputs.get("endpoint_ids"))}], "alias": inputs.get("alias")} + print(json.dumps(post("/endpoints/update_agent_name/", 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) diff --git a/integrations/cortex-xdr/scripts/set_user_role.py b/integrations/cortex-xdr/scripts/set_user_role.py new file mode 100644 index 0000000..9e1d13d --- /dev/null +++ b/integrations/cortex-xdr/scripts/set_user_role.py @@ -0,0 +1,63 @@ +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_raw(path, body): + base, headers = _client() + data = json.dumps(body).encode("utf-8") if body is not None else None + 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 post(path, request_data=None): + return post_raw(path, {"request_data": request_data} if request_data is not None else None) + + +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: + from datetime import datetime + 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", "{}")) + print(json.dumps(post("/rbac/set_user_role/", {"user_emails": csv(inputs.get("user_emails")), "role_name": inputs.get("role_name")}))) + + +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)