From d85a955e77ecf65cbe1d1c7d5648228b36a5dfab Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Thu, 25 Jun 2026 23:27:37 +0200 Subject: [PATCH] feat(crowdstrike): add CrowdStrike Falcon connector (IR enrichment, IOC, containment, RTR) Co-Authored-By: Claude Opus 4.8 --- integrations/crowdstrike/manifest.yaml | 228 ++++++++++++++++++ .../crowdstrike/scripts/contain_host.py | 47 ++++ .../crowdstrike/scripts/delete_custom_ioc.py | 48 ++++ .../scripts/device_online_state.py | 48 ++++ .../crowdstrike/scripts/get_device_details.py | 47 ++++ .../crowdstrike/scripts/ioc_device_count.py | 48 ++++ .../crowdstrike/scripts/lift_containment.py | 47 ++++ .../crowdstrike/scripts/list_host_groups.py | 52 ++++ .../crowdstrike/scripts/resolve_detection.py | 55 +++++ .../crowdstrike/scripts/rtr_kill_process.py | 56 +++++ .../crowdstrike/scripts/rtr_list_processes.py | 53 ++++ .../crowdstrike/scripts/rtr_run_command.py | 53 ++++ .../crowdstrike/scripts/rtr_run_script.py | 64 +++++ .../crowdstrike/scripts/search_detections.py | 55 +++++ .../crowdstrike/scripts/search_device.py | 67 +++++ .../scripts/spotlight_search_vulnerability.py | 54 +++++ .../crowdstrike/scripts/upload_custom_ioc.py | 60 +++++ 17 files changed, 1082 insertions(+) create mode 100644 integrations/crowdstrike/manifest.yaml create mode 100644 integrations/crowdstrike/scripts/contain_host.py create mode 100644 integrations/crowdstrike/scripts/delete_custom_ioc.py create mode 100644 integrations/crowdstrike/scripts/device_online_state.py create mode 100644 integrations/crowdstrike/scripts/get_device_details.py create mode 100644 integrations/crowdstrike/scripts/ioc_device_count.py create mode 100644 integrations/crowdstrike/scripts/lift_containment.py create mode 100644 integrations/crowdstrike/scripts/list_host_groups.py create mode 100644 integrations/crowdstrike/scripts/resolve_detection.py create mode 100644 integrations/crowdstrike/scripts/rtr_kill_process.py create mode 100644 integrations/crowdstrike/scripts/rtr_list_processes.py create mode 100644 integrations/crowdstrike/scripts/rtr_run_command.py create mode 100644 integrations/crowdstrike/scripts/rtr_run_script.py create mode 100644 integrations/crowdstrike/scripts/search_detections.py create mode 100644 integrations/crowdstrike/scripts/search_device.py create mode 100644 integrations/crowdstrike/scripts/spotlight_search_vulnerability.py create mode 100644 integrations/crowdstrike/scripts/upload_custom_ioc.py diff --git a/integrations/crowdstrike/manifest.yaml b/integrations/crowdstrike/manifest.yaml new file mode 100644 index 0000000..ec78232 --- /dev/null +++ b/integrations/crowdstrike/manifest.yaml @@ -0,0 +1,228 @@ +id: crowdstrike +name: CrowdStrike Falcon +version: 1.0.0 +description: "CrowdStrike Falcon (OAuth2 API) — device search & enrichment, detections, Spotlight vulnerabilities, custom IOC management, host containment, and Real Time Response." +changelog: "1.0.0 — Initial release: device/detection enrichment, Spotlight, IOC management, contain/lift, and RTR (run command/script, list/kill process)." +category: endpoint + +# Per-instance configuration. Scripts obtain an OAuth2 bearer token from +# {base_url}/oauth2/token using client_id/client_secret, then call the API. +config_schema: + properties: + base_url: + type: string + description: "CrowdStrike API base URL (region-specific), e.g. https://api.crowdstrike.com or https://api.eu-1.crowdstrike.com" + default: https://api.crowdstrike.com + client_id: + type: string + description: "Falcon API client ID (Support → API Clients and Keys)" + x-soar-sensitive: true + client_secret: + type: string + description: "Falcon API client secret" + x-soar-sensitive: true + required: + - client_id + - client_secret + +# Documented for reference; the bundled scripts perform the OAuth2 client- +# credentials exchange themselves and send Authorization: Bearer . +auth: + - id: oauth2 + type: oauth2_client_credentials + secret_field: client_secret + +commands: + # ── Enrichment / read ───────────────────────────────────────────────────── + - id: search_device + name: crowdstrike-search-device + description: "Search managed devices (hosts) by hostname, IP, platform, or status; returns full device details for matches." + risk: read + inputs_schema: + properties: + hostname: { type: string, description: "Hostname to match (exact)." } + ip: { type: string, description: "Local IP address to match." } + platform_name: { type: string, description: "Platform: Windows, Mac, or Linux." } + status: { type: string, description: "Containment status: normal, containment_pending, contained, or lift_containment_pending." } + limit: { type: number, description: "Max device IDs to resolve (default 50)." } + required: [] + outputs_schema: { properties: {} } + + - id: get_device_details + name: crowdstrike-get-device-details + description: "Retrieve full details for one or more devices by ID." + risk: read + inputs_schema: + properties: + ids: { type: string, description: "Comma-separated device IDs." } + required: [ids] + outputs_schema: { properties: {} } + + - id: device_online_state + name: crowdstrike-device-online-state + description: "Get the current online/offline state for one or more devices by ID." + risk: read + inputs_schema: + properties: + ids: { type: string, description: "Comma-separated device IDs." } + required: [ids] + outputs_schema: { properties: {} } + + - id: search_detections + name: crowdstrike-search-detections + description: "Search detections/alerts using an FQL filter; returns full alert entities for matches." + risk: read + inputs_schema: + properties: + filter: { type: string, description: "FQL filter (default \"product:'epp'+type:'ldt'\")." } + limit: { type: number, description: "Max alerts to return (default 50)." } + required: [] + outputs_schema: { properties: {} } + + - id: list_host_groups + name: crowdstrike-list-host-groups + description: "List host groups, optionally filtered by an FQL expression." + risk: read + inputs_schema: + properties: + filter: { type: string, description: "Optional FQL filter, e.g. \"name:*'*prod*'\"." } + limit: { type: number, description: "Max groups to return (default 50)." } + required: [] + outputs_schema: { properties: {} } + + - id: spotlight_search_vulnerability + name: crowdstrike-spotlight-search-vulnerability + description: "Search Falcon Spotlight vulnerabilities by FQL filter or CVE ID, with CVE facet detail." + risk: read + inputs_schema: + properties: + filter: { type: string, description: "FQL filter (defaults to \"status:'open'\")." } + cve_id: { type: string, description: "Comma-separated CVE IDs; overrides filter when provided." } + limit: { type: number, description: "Max vulnerabilities to return (default 50)." } + required: [] + outputs_schema: { properties: {} } + + - id: ioc_device_count + name: crowdstrike-ioc-device-count + description: "Count how many devices have observed a given indicator (IOC)." + risk: read + inputs_schema: + properties: + type: { type: string, description: "Indicator type: sha256, sha1, md5, domain, ipv4, or ipv6." } + value: { type: string, description: "Indicator value." } + required: [type, value] + outputs_schema: { properties: {} } + + # ── Actions (containment, detection workflow, IOC management) ────────────── + - id: contain_host + name: crowdstrike-contain-host + description: "Network-contain one or more hosts (isolate them while keeping Falcon connectivity)." + risk: destructive + inputs_schema: + properties: + ids: { type: string, description: "Comma-separated device IDs to contain." } + required: [ids] + outputs_schema: { properties: {} } + + - id: lift_containment + name: crowdstrike-lift-containment + description: "Lift network containment on one or more hosts (reconnect them)." + risk: destructive + inputs_schema: + properties: + ids: { type: string, description: "Comma-separated device IDs to release." } + required: [ids] + outputs_schema: { properties: {} } + + - id: resolve_detection + name: crowdstrike-resolve-detection + description: "Update detections/alerts: change status, append a comment, and/or assign to an analyst." + risk: destructive + inputs_schema: + properties: + ids: { type: string, description: "Comma-separated alert composite IDs." } + status: { type: string, description: "New status: new, in_progress, closed, or reopened." } + comment: { type: string, description: "Optional comment to append." } + assign_to_uuid: { type: string, description: "Optional analyst UUID to assign the detection to." } + required: [ids] + outputs_schema: { properties: {} } + + - id: upload_custom_ioc + name: crowdstrike-upload-custom-ioc + description: "Create a custom IOC indicator (allow/prevent/detect) applied globally." + risk: destructive + inputs_schema: + properties: + type: { type: string, description: "Indicator type: sha256, md5, domain, ipv4, or ipv6." } + value: { type: string, description: "Indicator value." } + action: { type: string, description: "Action: no_action, allow, prevent_no_ui, prevent, or detect." } + platforms: { type: string, description: "Comma-separated platforms: windows, mac, linux." } + severity: { type: string, description: "Optional severity: informational, low, medium, high, or critical." } + description: { type: string, description: "Optional description." } + source: { type: string, description: "Optional source label." } + required: [type, value, action, platforms] + outputs_schema: { properties: {} } + + - id: delete_custom_ioc + name: crowdstrike-delete-custom-ioc + description: "Delete one or more custom IOC indicators by ID." + risk: destructive + inputs_schema: + properties: + ids: { type: string, description: "Comma-separated IOC indicator IDs." } + required: [ids] + outputs_schema: { properties: {} } + + # ── Real Time Response ──────────────────────────────────────────────────── + - id: rtr_run_command + name: crowdstrike-rtr-run-command + description: "Open an RTR batch session and run a read-only command (e.g. ls, ps, cat) on the target hosts." + risk: destructive + inputs_schema: + properties: + host_ids: { type: string, description: "Comma-separated device IDs." } + command_type: { type: string, description: "Base command, e.g. ls, ps, cat, reg." } + full_command: { type: string, description: "Full command string, e.g. 'ls C:\\\\'." } + queue_offline: { type: string, description: "Queue for offline hosts: true or false (default false)." } + timeout: { type: number, description: "Command timeout in seconds (default 30)." } + required: [host_ids, command_type, full_command] + outputs_schema: { properties: {} } + + - id: rtr_run_script + name: crowdstrike-rtr-run-script + description: "Open an RTR batch session and run a script via runscript (cloud-stored by name or inline raw PowerShell)." + risk: destructive + inputs_schema: + properties: + host_ids: { type: string, description: "Comma-separated device IDs." } + script_name: { type: string, description: "Name of a cloud-stored RTR script (exactly one of script_name/raw)." } + raw: { type: string, description: "Inline raw PowerShell script (exactly one of script_name/raw)." } + queue_offline: { type: string, description: "Queue for offline hosts: true or false (default false)." } + timeout: { type: number, description: "Script timeout in seconds (default 30)." } + required: [host_ids] + outputs_schema: { properties: {} } + + - id: rtr_list_processes + name: crowdstrike-rtr-list-processes + description: "Open an RTR session on a single host and list running processes (ps)." + risk: destructive + inputs_schema: + properties: + host_id: { type: string, description: "A single device ID." } + queue_offline: { type: string, description: "Queue for offline hosts: true or false (default false)." } + timeout: { type: number, description: "Command timeout in seconds (default 30)." } + required: [host_id] + outputs_schema: { properties: {} } + + - id: rtr_kill_process + name: crowdstrike-rtr-kill-process + description: "Open an RTR session on a single host and kill one or more processes by PID." + risk: destructive + inputs_schema: + properties: + host_id: { type: string, description: "A single device ID." } + process_ids: { type: string, description: "Comma-separated process IDs (PIDs) to kill." } + queue_offline: { type: string, description: "Queue for offline hosts: true or false (default false)." } + timeout: { type: number, description: "Command timeout in seconds (default 30)." } + required: [host_id, process_ids] + outputs_schema: { properties: {} } diff --git a/integrations/crowdstrike/scripts/contain_host.py b/integrations/crowdstrike/scripts/contain_host.py new file mode 100644 index 0000000..359936e --- /dev/null +++ b/integrations/crowdstrike/scripts/contain_host.py @@ -0,0 +1,47 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + result = call("POST", "/devices/entities/devices-actions/v2", tok, params=[("action_name", "contain")], body={"ids": csv(INPUTS.get("ids"))}) + print(json.dumps(result)) + + +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/crowdstrike/scripts/delete_custom_ioc.py b/integrations/crowdstrike/scripts/delete_custom_ioc.py new file mode 100644 index 0000000..8b71728 --- /dev/null +++ b/integrations/crowdstrike/scripts/delete_custom_ioc.py @@ -0,0 +1,48 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + params = [("ids", x) for x in csv(INPUTS.get("ids"))] + result = call("DELETE", "/iocs/entities/indicators/v1", tok, params=params) + print(json.dumps(result)) + + +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/crowdstrike/scripts/device_online_state.py b/integrations/crowdstrike/scripts/device_online_state.py new file mode 100644 index 0000000..0fd800b --- /dev/null +++ b/integrations/crowdstrike/scripts/device_online_state.py @@ -0,0 +1,48 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + params = [("ids", x) for x in csv(INPUTS.get("ids"))] + result = call("GET", "/devices/entities/online-state/v1", tok, params=params) + print(json.dumps(result)) + + +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/crowdstrike/scripts/get_device_details.py b/integrations/crowdstrike/scripts/get_device_details.py new file mode 100644 index 0000000..5dad64c --- /dev/null +++ b/integrations/crowdstrike/scripts/get_device_details.py @@ -0,0 +1,47 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + result = call("POST", "/devices/entities/devices/v2", tok, body={"ids": csv(INPUTS.get("ids"))}) + print(json.dumps(result)) + + +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/crowdstrike/scripts/ioc_device_count.py b/integrations/crowdstrike/scripts/ioc_device_count.py new file mode 100644 index 0000000..bd48d6b --- /dev/null +++ b/integrations/crowdstrike/scripts/ioc_device_count.py @@ -0,0 +1,48 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + params = [("type", INPUTS.get("type", "")), ("value", INPUTS.get("value", ""))] + result = call("GET", "/indicators/aggregates/devices-count/v1", tok, params=params) + print(json.dumps(result)) + + +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/crowdstrike/scripts/lift_containment.py b/integrations/crowdstrike/scripts/lift_containment.py new file mode 100644 index 0000000..cc2762d --- /dev/null +++ b/integrations/crowdstrike/scripts/lift_containment.py @@ -0,0 +1,47 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + result = call("POST", "/devices/entities/devices-actions/v2", tok, params=[("action_name", "lift_containment")], body={"ids": csv(INPUTS.get("ids"))}) + print(json.dumps(result)) + + +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/crowdstrike/scripts/list_host_groups.py b/integrations/crowdstrike/scripts/list_host_groups.py new file mode 100644 index 0000000..bdcda01 --- /dev/null +++ b/integrations/crowdstrike/scripts/list_host_groups.py @@ -0,0 +1,52 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + limit = INPUTS.get("limit", 50) + params = [] + if INPUTS.get("filter"): + params.append(("filter", INPUTS["filter"])) + params.append(("limit", limit)) + result = call("GET", "/devices/combined/host-groups/v1", tok, params=params) + print(json.dumps(result)) + + +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/crowdstrike/scripts/resolve_detection.py b/integrations/crowdstrike/scripts/resolve_detection.py new file mode 100644 index 0000000..a8a50ef --- /dev/null +++ b/integrations/crowdstrike/scripts/resolve_detection.py @@ -0,0 +1,55 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + action_params = [] + if INPUTS.get("status"): + action_params.append({"name": "update_status", "value": str(INPUTS["status"])}) + if INPUTS.get("comment"): + action_params.append({"name": "append_comment", "value": str(INPUTS["comment"])}) + if INPUTS.get("assign_to_uuid"): + action_params.append({"name": "assign_to_uuid", "value": str(INPUTS["assign_to_uuid"])}) + body = {"action_parameters": action_params, "composite_ids": csv(INPUTS.get("ids"))} + result = call("PATCH", "/alerts/entities/alerts/v3", tok, body=body) + print(json.dumps(result)) + + +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/crowdstrike/scripts/rtr_kill_process.py b/integrations/crowdstrike/scripts/rtr_kill_process.py new file mode 100644 index 0000000..1c46031 --- /dev/null +++ b/integrations/crowdstrike/scripts/rtr_kill_process.py @@ -0,0 +1,56 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + queue_offline = str(INPUTS.get("queue_offline", "")).lower() == "true" + timeout = INPUTS.get("timeout", 30) + init = call("POST", "/real-time-response/combined/batch-init-session/v1", tok, + body={"host_ids": [INPUTS.get("host_id")], "queue_offline": queue_offline}) + batch_id = init.get("batch_id") + results = [] + for pid in csv(INPUTS.get("process_ids")): + body = {"base_command": "kill", "batch_id": batch_id, "command_string": "kill " + pid} + resp = call("POST", "/real-time-response/combined/batch-active-responder-command/v1", tok, params=[("timeout", timeout)], body=body) + results.append(resp) + print(json.dumps({"results": results})) + + +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/crowdstrike/scripts/rtr_list_processes.py b/integrations/crowdstrike/scripts/rtr_list_processes.py new file mode 100644 index 0000000..c9ffa35 --- /dev/null +++ b/integrations/crowdstrike/scripts/rtr_list_processes.py @@ -0,0 +1,53 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + queue_offline = str(INPUTS.get("queue_offline", "")).lower() == "true" + timeout = INPUTS.get("timeout", 30) + init = call("POST", "/real-time-response/combined/batch-init-session/v1", tok, + body={"host_ids": [INPUTS.get("host_id")], "queue_offline": queue_offline}) + batch_id = init.get("batch_id") + body = {"base_command": "ps", "batch_id": batch_id, "command_string": "ps"} + result = call("POST", "/real-time-response/combined/batch-active-responder-command/v1", tok, params=[("timeout", timeout)], body=body) + print(json.dumps(result)) + + +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/crowdstrike/scripts/rtr_run_command.py b/integrations/crowdstrike/scripts/rtr_run_command.py new file mode 100644 index 0000000..e8dae3f --- /dev/null +++ b/integrations/crowdstrike/scripts/rtr_run_command.py @@ -0,0 +1,53 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + queue_offline = str(INPUTS.get("queue_offline", "")).lower() == "true" + timeout = INPUTS.get("timeout", 30) + init = call("POST", "/real-time-response/combined/batch-init-session/v1", tok, + body={"host_ids": csv(INPUTS.get("host_ids")), "queue_offline": queue_offline}) + batch_id = init.get("batch_id") + body = {"base_command": INPUTS.get("command_type"), "batch_id": batch_id, "command_string": INPUTS.get("full_command")} + result = call("POST", "/real-time-response/combined/batch-command/v1", tok, params=[("timeout", timeout)], body=body) + print(json.dumps(result)) + + +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/crowdstrike/scripts/rtr_run_script.py b/integrations/crowdstrike/scripts/rtr_run_script.py new file mode 100644 index 0000000..0e04d1b --- /dev/null +++ b/integrations/crowdstrike/scripts/rtr_run_script.py @@ -0,0 +1,64 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + script_name = INPUTS.get("script_name") + raw = INPUTS.get("raw") + if bool(script_name) == bool(raw): + print(json.dumps({"error": "Provide exactly one of script_name or raw."})) + sys.exit(1) + timeout = INPUTS.get("timeout", 30) + if script_name: + full = "runscript -CloudFile=" + str(script_name) + else: + full = "runscript -Raw=```" + str(raw) + "```" + full += " -Timeout=" + str(timeout) + + tok = token() + queue_offline = str(INPUTS.get("queue_offline", "")).lower() == "true" + init = call("POST", "/real-time-response/combined/batch-init-session/v1", tok, + body={"host_ids": csv(INPUTS.get("host_ids")), "queue_offline": queue_offline}) + batch_id = init.get("batch_id") + body = {"base_command": "runscript", "batch_id": batch_id, "command_string": full} + result = call("POST", "/real-time-response/combined/batch-admin-command/v1", tok, params=[("timeout", timeout)], body=body) + print(json.dumps(result)) + + +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/crowdstrike/scripts/search_detections.py b/integrations/crowdstrike/scripts/search_detections.py new file mode 100644 index 0000000..b00e512 --- /dev/null +++ b/integrations/crowdstrike/scripts/search_detections.py @@ -0,0 +1,55 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + fql = INPUTS.get("filter") or "product:'epp'+type:'ldt'" + limit = INPUTS.get("limit", 50) + params = [("filter", fql), ("limit", limit), ("sort", "created_timestamp.desc")] + found = call("GET", "/alerts/queries/alerts/v2", tok, params=params) + ids = found.get("resources", []) or [] + if ids: + result = call("POST", "/alerts/entities/alerts/v2", tok, body={"composite_ids": ids}) + else: + result = {"resources": []} + print(json.dumps(result)) + + +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/crowdstrike/scripts/search_device.py b/integrations/crowdstrike/scripts/search_device.py new file mode 100644 index 0000000..d3abf9b --- /dev/null +++ b/integrations/crowdstrike/scripts/search_device.py @@ -0,0 +1,67 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + clauses = [] + if INPUTS.get("hostname"): + clauses.append("hostname:'" + str(INPUTS["hostname"]) + "'") + if INPUTS.get("ip"): + clauses.append("local_ip:'" + str(INPUTS["ip"]) + "'") + if INPUTS.get("platform_name"): + clauses.append("platform_name:'" + str(INPUTS["platform_name"]) + "'") + if INPUTS.get("status"): + clauses.append("status:'" + str(INPUTS["status"]) + "'") + fql = "+".join(clauses) + limit = INPUTS.get("limit", 50) + params = [] + if fql: + params.append(("filter", fql)) + params.append(("limit", limit)) + found = call("GET", "/devices/queries/devices/v1", tok, params=params) + ids = found.get("resources", []) or [] + if ids: + result = call("POST", "/devices/entities/devices/v2", tok, body={"ids": ids}) + else: + result = {"resources": []} + print(json.dumps(result)) + + +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/crowdstrike/scripts/spotlight_search_vulnerability.py b/integrations/crowdstrike/scripts/spotlight_search_vulnerability.py new file mode 100644 index 0000000..ef7e297 --- /dev/null +++ b/integrations/crowdstrike/scripts/spotlight_search_vulnerability.py @@ -0,0 +1,54 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + cves = csv(INPUTS.get("cve_id")) + if cves: + fql = "cve.id:[" + ",".join("'" + c + "'" for c in cves) + "]" + else: + fql = INPUTS.get("filter") or "status:'open'" + limit = INPUTS.get("limit", 50) + params = [("filter", fql), ("facet", "cve"), ("limit", limit)] + result = call("GET", "/spotlight/combined/vulnerabilities/v1", tok, params=params) + print(json.dumps(result)) + + +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/crowdstrike/scripts/upload_custom_ioc.py b/integrations/crowdstrike/scripts/upload_custom_ioc.py new file mode 100644 index 0000000..5bce8ff --- /dev/null +++ b/integrations/crowdstrike/scripts/upload_custom_ioc.py @@ -0,0 +1,60 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) +INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) +BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/") + + +def token(): + data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode() + req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read()).get("access_token", "") + + +def call(method, path, tok, params=None, body=None): + url = BASE + path + if params: + # params is a list of (key, value) tuples to allow repeated keys (e.g. ids) + url += "?" + urllib.parse.urlencode(params) + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + 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(): + tok = token() + indicator = { + "type": INPUTS.get("type"), + "value": INPUTS.get("value"), + "action": INPUTS.get("action"), + "platforms": csv(INPUTS.get("platforms")), + "applied_globally": True, + } + if INPUTS.get("severity"): + indicator["severity"] = INPUTS["severity"] + if INPUTS.get("description"): + indicator["description"] = INPUTS["description"] + if INPUTS.get("source"): + indicator["source"] = INPUTS["source"] + result = call("POST", "/iocs/entities/indicators/v1", tok, body={"indicators": [indicator]}) + print(json.dumps(result)) + + +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)