feat(crowdstrike): expand to full command set (81 commands)

Adds device/IOC/process enrichment, host groups, cases, RTR
scripts/files/responders, ML/IOA exclusions, quarantine, Spotlight/CVE,
ODS scans, CSPM, users, IOA rules, CNAPP, detection resolve, workflows,
and identity (GraphQL). OAuth2 client-credentials, code-first scripts.

Excludes XSOAR-only plumbing (fetch/mirroring), deprecated legacy
commands, and binary/long-poll flows (retrieve-file, ngsiem search).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
2026-06-25 23:49:40 +02:00
parent d85a955e77
commit 17faa01103
66 changed files with 4572 additions and 0 deletions
@@ -0,0 +1,49 @@
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()
id_ = INPUTS.get("id", "")
tags = INPUTS.get("tags", "")
result = call("POST", "/cases/entities/case-tags/v1", tok, body={"id": id_, "tags": csv(tags)})
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)
@@ -0,0 +1,58 @@
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()
host_group_id = INPUTS.get("host_group_id", "")
host_ids = INPUTS.get("host_ids", "")
result = call(
"POST",
"/devices/entities/host-group-actions/v1",
tok,
params=[("action_name", "add-hosts")],
body={
"action_parameters": [{"name": "filter", "value": "(device_id:" + str(csv(host_ids)) + ")"}],
"ids": [host_group_id],
},
)
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)
@@ -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()
ids = INPUTS.get("ids", "")
action = INPUTS.get("action", "")
comment = INPUTS.get("comment", "")
body = {
"ids": csv(ids),
"action": action,
"comment": comment,
}
result = call("PATCH", "/quarantine/entities/quarantined-files/v1", 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)
@@ -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()
parsed = json.loads(INPUTS.get("multiple_indicators_json", "[]"))
result = call("POST", "/iocs/entities/indicators/v1", tok, body={"indicators": parsed})
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)
@@ -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()
name = INPUTS.get("name", "")
group_type = INPUTS.get("group_type", "")
description = INPUTS.get("description", "")
assignment_rule = INPUTS.get("assignment_rule", "")
res = {"name": name, "group_type": group_type}
if description:
res["description"] = description
if assignment_rule:
res["assignment_rule"] = assignment_rule
result = call("POST", "/devices/entities/host-groups/v1", tok, body={"resources": [res]})
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)
@@ -0,0 +1,73 @@
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()
exclusion_name = INPUTS.get("exclusion_name", "")
pattern_id = INPUTS.get("pattern_id", "")
cl_regex = INPUTS.get("cl_regex", "")
ifn_regex = INPUTS.get("ifn_regex", "")
pattern_name = INPUTS.get("pattern_name", "")
comment = INPUTS.get("comment", "")
description = INPUTS.get("description", "")
groups = INPUTS.get("groups", "")
detection_json = INPUTS.get("detection_json", "")
body = {
"name": exclusion_name,
"pattern_id": pattern_id,
"cl_regex": cl_regex,
"ifn_regex": ifn_regex,
"groups": csv(groups or "all"),
}
if pattern_name:
body["pattern_name"] = pattern_name
if comment:
body["comment"] = comment
if description:
body["description"] = description
if detection_json:
body["detection_json"] = detection_json
result = call("POST", "/policy/entities/ioa-exclusions/v1", 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)
@@ -0,0 +1,58 @@
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()
value = INPUTS.get("value", "")
excluded_from = INPUTS.get("excluded_from", "")
comment = INPUTS.get("comment", "")
groups = INPUTS.get("groups", "")
body = {
"value": value,
"excluded_from": csv(excluded_from),
"groups": csv(groups or "all"),
}
if comment:
body["comment"] = comment
result = call("POST", "/policy/entities/ml-exclusions/v1", 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)
@@ -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()
policy_ids = INPUTS.get("policy_ids", "")
result = call("GET", "/settings/entities/policy-details/v1", tok, params=[("ids", x) for x in csv(policy_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)
@@ -0,0 +1,58 @@
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()
policy_id = INPUTS.get("policy_id", "")
cloud_platform = INPUTS.get("cloud_platform", "aws")
service = INPUTS.get("service", "")
limit = INPUTS.get("limit", "")
params = [("policy-id", policy_id), ("cloud-platform", cloud_platform)]
if service:
params.append(("service", service))
if limit:
params.append(("limit", limit))
result = call("GET", "/settings/entities/policy/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)
@@ -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()
policy_id = INPUTS.get("policy_id", "")
account_id = INPUTS.get("account_id", "")
enabled = INPUTS.get("enabled", "true")
regions = INPUTS.get("regions", "")
severity = INPUTS.get("severity", "")
tag_excluded = INPUTS.get("tag_excluded", "")
res = {
"policy_id": int(policy_id),
"enabled": (str(enabled).lower() == "true"),
}
if account_id:
res["account_id"] = account_id
if severity:
res["severity"] = severity
if regions:
res["regions"] = csv(regions)
if tag_excluded:
res["tag_excluded"] = (str(tag_excluded).lower() == "true")
result = call("PATCH", "/settings/entities/policy/v1", tok, body={"resources": [res]})
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)
+49
View File
@@ -0,0 +1,49 @@
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()
cve = INPUTS.get("cve", "")
filt = "cve.id:['" + "','".join(csv(cve)) + "']"
result = call("GET", "/spotlight/combined/vulnerabilities/v1", tok, params=[("filter", filt), ("facet", "cve")])
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)
@@ -0,0 +1,49 @@
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()
id_ = INPUTS.get("id", "")
tag = INPUTS.get("tag", "")
result = call("DELETE", "/cases/entities/case-tags/v1", tok, params=[("id", id_), ("tag", tag)])
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)
@@ -0,0 +1,49 @@
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()
file_id = INPUTS.get("file_id", "")
result = call("DELETE", "/real-time-response/entities/put-files/v1", tok,
params=[("ids", file_id)])
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)
@@ -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()
host_group_id = INPUTS.get("host_group_id", "")
result = call("DELETE", "/devices/entities/host-groups/v1", tok, params=[("ids", x) for x in csv(host_group_id)])
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)
@@ -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()
ids = INPUTS.get("ids", "")
result = call("DELETE", "/policy/entities/ioa-exclusions/v1", tok, params=[("ids", x) for x in csv(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)
@@ -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()
ids = INPUTS.get("ids", "")
result = call("DELETE", "/policy/entities/ml-exclusions/v1", tok, params=[("ids", x) for x in csv(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)
@@ -0,0 +1,49 @@
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()
script_id = INPUTS.get("script_id", "")
result = call("DELETE", "/real-time-response/entities/scripts/v1", tok,
params=[("ids", script_id)])
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)
@@ -0,0 +1,49 @@
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()
type_ = INPUTS.get("type", "")
value = INPUTS.get("value", "")
result = call("GET", "/indicators/queries/devices/v1", tok, params=[("type", type_), ("value", value)])
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)
@@ -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()
parts = []
if INPUTS.get("id"):
parts.append("device_id:'" + INPUTS.get("id") + "'")
if INPUTS.get("ip"):
parts.append("local_ip:'" + INPUTS.get("ip") + "'")
if INPUTS.get("hostname"):
parts.append("hostname:'" + INPUTS.get("hostname") + "'")
filt = "+".join(parts)
query = call("GET", "/devices/queries/devices/v1", tok, params=[("filter", filt), ("limit", "50")])
ids = query.get("resources", [])
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)
@@ -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()
ioc_id = INPUTS.get("ioc_id", "")
result = call("GET", "/iocs/entities/indicators/v1", tok, params=[("ids", ioc_id)])
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)
@@ -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()
id_ = INPUTS.get("id", "")
result = call("POST", "/cases/entities/cases/v2", tok, body={"ids": [id_]})
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)
@@ -0,0 +1,49 @@
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()
file_id = INPUTS.get("file_id", "")
result = call("GET", "/real-time-response/entities/put-files/v2", tok,
params=[("ids", x) for x in csv(file_id)])
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)
@@ -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()
rule_ids = INPUTS.get("rule_ids", "")
result = call("GET", "/ioarules/entities/rules/v1", tok, params=[("ids", x) for x in csv(rule_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)
@@ -0,0 +1,49 @@
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()
script_id = INPUTS.get("script_id", "")
result = call("GET", "/real-time-response/entities/scripts/v2", tok,
params=[("ids", x) for x in csv(script_id)])
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)
@@ -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()
ids_input = INPUTS.get("ids", "")
if ids_input:
result = call("POST", "/cases/entities/cases/v2", tok, body={"ids": csv(ids_input)})
else:
query = call("GET", "/cases/queries/cases/v1", tok, params=[("sort", "created_timestamp.asc"), ("limit", "100")])
resources = query.get("resources", [])
result = call("POST", "/cases/entities/cases/v2", tok, body={"ids": 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)
@@ -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()
filter_ = INPUTS.get("filter", "")
params = [("offset", "0"), ("limit", "100")]
if filter_:
params.append(("filter", filter_))
result = call("GET", "/container-security/combined/container-alerts/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)
@@ -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()
ids_input = INPUTS.get("ids", "")
if ids_input:
comp = csv(ids_input)
else:
fetch_query = INPUTS.get("fetch_query", "")
params = [("filter", fetch_query)] if fetch_query else []
query = call("GET", "/alerts/queries/alerts/v2", tok, params=params)
comp = query.get("resources", [])
result = call("POST", "/alerts/entities/alerts/v2", tok, body={"composite_ids": comp})
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)
@@ -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("GET", "/real-time-response/entities/put-files/v2", tok)
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)
@@ -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()
host_id = INPUTS.get("host_id", "")
session_id = INPUTS.get("session_id", "")
if not session_id:
init = call("POST", "/real-time-response/entities/sessions/v1", tok,
body={"device_id": host_id})
session_id = init.get("resources", [{}])[0].get("session_id", "")
result = call("GET", "/real-time-response/entities/file/v2", tok,
params=[("session_id", session_id)])
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)
@@ -0,0 +1,63 @@
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()
host_group_id = INPUTS.get("host_group_id", "")
filter_ = INPUTS.get("filter", "")
limit = INPUTS.get("limit", "50")
offset = INPUTS.get("offset", "")
sort = INPUTS.get("sort", "")
params = []
if host_group_id:
params.append(("id", host_group_id))
if filter_:
params.append(("filter", filter_))
if limit:
params.append(("limit", limit))
if offset:
params.append(("offset", offset))
if sort:
params.append(("sort", sort))
result = call("GET", "/devices/combined/host-group-members/v1", tok, params=params if params else None)
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)
@@ -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()
q = '{ entities(types: [%s], first: %s) { nodes { primaryDisplayName secondaryDisplayName riskScore riskScoreSeverity riskFactors { type severity } ... on UserEntity { emailAddresses } } } }' % (INPUTS.get("type", "USER"), str(INPUTS.get("limit", "50")))
result = call("POST", "/identity-protection/combined/graphql/v1", tok, body={"query": q})
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)
@@ -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()
ids = INPUTS.get("ids", "")
filter_ = INPUTS.get("filter", "")
limit = INPUTS.get("limit", "50")
offset = INPUTS.get("offset", "")
if ids:
q_ids = csv(ids)
else:
params = []
if filter_:
params.append(("filter", filter_))
params.append(("limit", limit))
if offset:
params.append(("offset", offset))
resp = call("GET", "/quarantine/queries/quarantined-files/v1", tok, params=params)
q_ids = resp.get("resources", [])
if q_ids:
result = call("POST", "/quarantine/entities/quarantined-files/GET/v1", tok, body={"ids": q_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)
@@ -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("GET", "/real-time-response/entities/scripts/v2", tok)
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)
@@ -0,0 +1,61 @@
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()
id_ = INPUTS.get("id", "")
filter_ = INPUTS.get("filter", "")
offset = INPUTS.get("offset", "0")
limit = INPUTS.get("limit", "50")
if id_:
u_ids = csv(id_)
else:
params = [("offset", offset), ("limit", limit), ("sort", "uid")]
if filter_:
params.append(("filter", filter_))
query_res = call("GET", "/user-management/queries/users/v1", tok, params=params)
u_ids = query_res.get("resources", [])
result = call("POST", "/user-management/entities/users/GET/v1", tok, body={"ids": u_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)
@@ -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()
filter_ = INPUTS.get("filter", "")
offset = INPUTS.get("offset", "0")
limit = INPUTS.get("limit", "50")
sort = INPUTS.get("sort", "")
params = []
if filter_:
params.append(("filter", filter_))
if offset:
params.append(("offset", offset))
if limit:
params.append(("limit", limit))
if sort:
params.append(("sort", sort))
result = call("GET", "/workflows/combined/definitions/v1", tok, params=params or None)
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)
@@ -0,0 +1,49 @@
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()
ids = INPUTS.get("ids", "")
params = [("ids", x) for x in csv(ids)]
result = call("GET", "/workflows/entities/execution-results/v1", tok, params=params or None)
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)
@@ -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()
filter_ = INPUTS.get("filter", "")
offset = INPUTS.get("offset", "0")
limit = INPUTS.get("limit", "50")
sort = INPUTS.get("sort", "")
params = []
if filter_:
params.append(("filter", filter_))
if offset:
params.append(("offset", offset))
if limit:
params.append(("limit", limit))
if sort:
params.append(("sort", sort))
result = call("GET", "/workflows/combined/executions/v1", tok, params=params or None)
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)
@@ -0,0 +1,68 @@
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()
hosts = INPUTS.get("hosts", "")
host_groups = INPUTS.get("host_groups", "")
file_paths = INPUTS.get("file_paths", "")
scan_inclusions = INPUTS.get("scan_inclusions", "")
description = INPUTS.get("description", "")
quarantine = INPUTS.get("quarantine", "")
body = {}
if hosts:
body["hosts"] = csv(hosts)
if host_groups:
body["host_groups"] = csv(host_groups)
if file_paths:
body["file_paths"] = csv(file_paths)
if scan_inclusions:
body["scan_inclusions"] = csv(scan_inclusions)
if description:
body["description"] = description
if quarantine:
body["quarantine"] = (str(quarantine).lower() == "true")
result = call("POST", "/ods/entities/scans/v1", 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)
@@ -0,0 +1,77 @@
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()
host_groups = INPUTS.get("host_groups", "")
file_paths = INPUTS.get("file_paths", "")
scan_inclusions = INPUTS.get("scan_inclusions", "")
description = INPUTS.get("description", "")
schedule_start_timestamp = INPUTS.get("schedule_start_timestamp", "")
schedule_interval = INPUTS.get("schedule_interval", "")
interval_map = {
"never": 0,
"daily": 1,
"weekly": 7,
"every other week": 14,
"every four weeks": 28,
"monthly": 30,
}
body = {
"host_groups": csv(host_groups),
"schedule": {
"interval": interval_map.get(schedule_interval.lower(), 0),
"start_timestamp": schedule_start_timestamp,
},
}
if file_paths:
body["file_paths"] = csv(file_paths)
if scan_inclusions:
body["scan_inclusions"] = csv(scan_inclusions)
if description:
body["description"] = description
result = call("POST", "/ods/entities/scheduled-scans/v1", 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)
@@ -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()
ids = INPUTS.get("ids", "")
result = call("DELETE", "/ods/entities/scheduled-scans/v1", tok, params=[("ids", x) for x in csv(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)
@@ -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()
file_ids = INPUTS.get("file_ids", "")
filter_ = INPUTS.get("filter", "")
limit = INPUTS.get("limit", "")
offset = INPUTS.get("offset", "")
if file_ids:
result = call("GET", "/ods/entities/malicious-files/v1", tok, params=[("ids", x) for x in csv(file_ids)])
else:
params = []
if filter_:
params.append(("filter", filter_))
if limit:
params.append(("limit", limit))
if offset:
params.append(("offset", offset))
query_res = call("GET", "/ods/queries/malicious-files/v1", tok, params=params or None)
resources = query_res.get("resources", [])
if resources:
result = call("GET", "/ods/entities/malicious-files/v1", tok, params=[("ids", x) for x in resources])
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)
@@ -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()
ids = INPUTS.get("ids", "")
filter_ = INPUTS.get("filter", "")
limit = INPUTS.get("limit", "")
offset = INPUTS.get("offset", "")
if ids:
result = call("GET", "/ods/entities/scans/v1", tok, params=[("ids", x) for x in csv(ids)])
else:
params = []
if filter_:
params.append(("filter", filter_))
if limit:
params.append(("limit", limit))
if offset:
params.append(("offset", offset))
query_res = call("GET", "/ods/queries/scans/v1", tok, params=params or None)
resources = query_res.get("resources", [])
if resources:
result = call("GET", "/ods/entities/scans/v1", tok, params=[("ids", x) for x in resources])
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)
@@ -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()
ids = INPUTS.get("ids", "")
filter_ = INPUTS.get("filter", "")
limit = INPUTS.get("limit", "")
offset = INPUTS.get("offset", "")
if ids:
result = call("GET", "/ods/entities/scan-hosts/v1", tok, params=[("ids", x) for x in csv(ids)])
else:
params = []
if filter_:
params.append(("filter", filter_))
if limit:
params.append(("limit", limit))
if offset:
params.append(("offset", offset))
query_res = call("GET", "/ods/queries/scan-hosts/v1", tok, params=params or None)
resources = query_res.get("resources", [])
if resources:
result = call("GET", "/ods/entities/scan-hosts/v1", tok, params=[("ids", x) for x in resources])
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)
@@ -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()
ids = INPUTS.get("ids", "")
filter_ = INPUTS.get("filter", "")
limit = INPUTS.get("limit", "")
offset = INPUTS.get("offset", "")
if ids:
result = call("GET", "/ods/entities/scheduled-scans/v1", tok, params=[("ids", x) for x in csv(ids)])
else:
params = []
if filter_:
params.append(("filter", filter_))
if limit:
params.append(("limit", limit))
if offset:
params.append(("offset", offset))
query_res = call("GET", "/ods/queries/scheduled-scans/v1", tok, params=params or None)
resources = query_res.get("resources", [])
if resources:
result = call("GET", "/ods/entities/scheduled-scans/v1", tok, params=[("ids", x) for x in resources])
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)
@@ -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()
ids = INPUTS.get("ids", "")
result = call("GET", "/processes/entities/processes/v1", tok, params=[("ids", x) for x in csv(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)
@@ -0,0 +1,50 @@
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()
type_ = INPUTS.get("type", "")
value = INPUTS.get("value", "")
device_id = INPUTS.get("device_id", "")
result = call("GET", "/indicators/queries/processes/v1", tok, params=[("type", type_), ("value", value), ("device_id", device_id)])
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)
@@ -0,0 +1,49 @@
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()
host_id = INPUTS.get("host_id", "")
result = call("POST", "/real-time-response/entities/refresh-session/v1", tok,
body={"device_id": host_id})
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)
@@ -0,0 +1,58 @@
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()
host_group_id = INPUTS.get("host_group_id", "")
host_ids = INPUTS.get("host_ids", "")
result = call(
"POST",
"/devices/entities/host-group-actions/v1",
tok,
params=[("action_name", "remove-hosts")],
body={
"action_parameters": [{"name": "filter", "value": "(device_id:" + str(csv(host_ids)) + ")"}],
"ids": [host_group_id],
},
)
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)
@@ -0,0 +1,59 @@
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()
id_ = INPUTS.get("id", "")
fields = {}
if INPUTS.get("status"):
fields["status"] = INPUTS.get("status")
if INPUTS.get("name"):
fields["name"] = INPUTS.get("name")
if INPUTS.get("description"):
fields["description"] = INPUTS.get("description")
if INPUTS.get("assigned_to_uuid"):
fields["assigned_to_user_uuid"] = INPUTS.get("assigned_to_uuid")
if INPUTS.get("severity"):
fields["severity"] = int(INPUTS.get("severity"))
result = call("PATCH", "/cases/entities/cases/v2", tok, body={"id": id_, "fields": fields})
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)
@@ -0,0 +1,72 @@
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()
ids = INPUTS.get("ids", "")
update_status = INPUTS.get("update_status", "")
assign_to_uuid = INPUTS.get("assign_to_uuid", "")
append_comment = INPUTS.get("append_comment", "")
add_tag = INPUTS.get("add_tag", "")
remove_tag = INPUTS.get("remove_tag", "")
unassign = INPUTS.get("unassign", "")
show_in_ui = INPUTS.get("show_in_ui", "")
action_parameters = []
for name, value in [
("update_status", update_status),
("assign_to_uuid", assign_to_uuid),
("append_comment", append_comment),
("add_tag", add_tag),
("remove_tag", remove_tag),
("unassign", unassign),
("show_in_ui", show_in_ui),
]:
if value:
action_parameters.append({"name": name, "value": value})
result = call("PATCH", "/alerts/entities/alerts/v3", tok, body={
"action_parameters": action_parameters,
"composite_ids": csv(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)
@@ -0,0 +1,72 @@
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()
ids = INPUTS.get("ids", "")
update_status = INPUTS.get("update_status", "")
assign_to_uuid = INPUTS.get("assign_to_uuid", "")
append_comment = INPUTS.get("append_comment", "")
add_tag = INPUTS.get("add_tag", "")
remove_tag = INPUTS.get("remove_tag", "")
unassign = INPUTS.get("unassign", "")
show_in_ui = INPUTS.get("show_in_ui", "")
action_parameters = []
for name, value in [
("update_status", update_status),
("assign_to_uuid", assign_to_uuid),
("append_comment", append_comment),
("add_tag", add_tag),
("remove_tag", remove_tag),
("unassign", unassign),
("show_in_ui", show_in_ui),
]:
if value:
action_parameters.append({"name": name, "value": value})
result = call("PATCH", "/alerts/entities/alerts/v3", tok, body={
"action_parameters": action_parameters,
"composite_ids": csv(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)
@@ -0,0 +1,57 @@
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()
host_id = INPUTS.get("host_id", "")
qoff = str(INPUTS.get("queue_offline", "false")).lower() == "true"
timeout = INPUTS.get("timeout", "30")
init = call("POST", "/real-time-response/combined/batch-init-session/v1", tok,
body={"host_ids": [host_id], "queue_offline": qoff})
batch_id = init.get("batch_id")
result = call("POST", "/real-time-response/combined/batch-active-responder-command/v1", tok,
params=[("timeout", timeout)],
body={"base_command": "netstat", "batch_id": batch_id, "command_string": "netstat"})
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)
@@ -0,0 +1,59 @@
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()
host_ids = INPUTS.get("host_ids", "")
qoff = str(INPUTS.get("queue_offline", "false")).lower() == "true"
timeout = INPUTS.get("timeout", "30")
init = call("POST", "/real-time-response/combined/batch-init-session/v1", tok,
body={"host_ids": csv(host_ids), "queue_offline": qoff})
batch_id = init.get("batch_id")
command_string = "runscript -Raw=" + chr(96) * 3 + "schtasks /query /fo LIST /v" + chr(96) * 3
result = call("POST", "/real-time-response/combined/batch-admin-command/v1", tok,
params=[("timeout", timeout)],
body={"base_command": "runscript", "batch_id": batch_id, "command_string": command_string})
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)
@@ -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():
tok = token()
host_ids = INPUTS.get("host_ids", "")
file_path = INPUTS.get("file_path", "")
os_type = INPUTS.get("os", "Windows")
qoff = str(INPUTS.get("queue_offline", "false")).lower() == "true"
timeout = INPUTS.get("timeout", "30")
init = call("POST", "/real-time-response/combined/batch-init-session/v1", tok,
body={"host_ids": csv(host_ids), "queue_offline": qoff})
batch_id = init.get("batch_id")
if os_type == "Windows":
full = "rm '" + file_path + "' --force"
else:
full = "rm '" + file_path + "' -r -d"
result = call("POST", "/real-time-response/combined/batch-active-responder-command/v1", tok,
params=[("timeout", timeout)],
body={"base_command": "rm", "batch_id": batch_id, "command_string": full})
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)
@@ -0,0 +1,76 @@
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()
types = INPUTS.get("types", "")
values = INPUTS.get("values", "")
sources = INPUTS.get("sources", "")
expiration = INPUTS.get("expiration", "")
limit = INPUTS.get("limit", "50")
sort = INPUTS.get("sort", "")
offset = INPUTS.get("offset", "")
filter_parts = []
if types:
filter_parts.append("type:" + types)
if values:
filter_parts.append("value:" + values)
if sources:
filter_parts.append("source:" + sources)
if expiration:
filter_parts.append("expiration:\"" + expiration + "\"")
filt = "+".join(filter_parts)
params = []
if filt:
params.append(("filter", filt))
if sort:
params.append(("sort", sort))
if offset:
params.append(("offset", offset))
if limit:
params.append(("limit", limit))
result = call("GET", "/iocs/combined/indicator/v1", tok, params=params if params else None)
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)
@@ -0,0 +1,77 @@
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()
ids = INPUTS.get("ids", "")
name = INPUTS.get("name", "")
filter_ = INPUTS.get("filter", "")
limit = INPUTS.get("limit", "")
offset = INPUTS.get("offset", "")
if ids:
ex_ids = csv(ids)
elif name:
params = [("filter", "name:~'" + name + "'")]
if limit:
params.append(("limit", limit))
if offset:
params.append(("offset", offset))
resp = call("GET", "/policy/queries/ioa-exclusions/v1", tok, params=params)
ex_ids = resp.get("resources", [])
else:
params = []
if filter_:
params.append(("filter", filter_))
if limit:
params.append(("limit", limit))
if offset:
params.append(("offset", offset))
resp = call("GET", "/policy/queries/ioa-exclusions/v1", tok, params=params if params else None)
ex_ids = resp.get("resources", [])
if ex_ids:
result = call("GET", "/policy/entities/ioa-exclusions/v1", tok, params=[("ids", x) for x in ex_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)
@@ -0,0 +1,82 @@
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()
ids = INPUTS.get("ids", "")
value = INPUTS.get("value", "")
filter_ = INPUTS.get("filter", "")
limit = INPUTS.get("limit", "")
offset = INPUTS.get("offset", "")
sort = INPUTS.get("sort", "")
if ids:
ex_ids = csv(ids)
elif value:
params = [("filter", "value:'" + value + "'")]
if limit:
params.append(("limit", limit))
if offset:
params.append(("offset", offset))
if sort:
params.append(("sort", sort))
resp = call("GET", "/policy/queries/ml-exclusions/v1", tok, params=params)
ex_ids = resp.get("resources", [])
else:
params = []
if filter_:
params.append(("filter", filter_))
if limit:
params.append(("limit", limit))
if offset:
params.append(("offset", offset))
if sort:
params.append(("sort", sort))
resp = call("GET", "/policy/queries/ml-exclusions/v1", tok, params=params if params else None)
ex_ids = resp.get("resources", [])
if ex_ids:
result = call("GET", "/policy/entities/ml-exclusions/v1", tok, params=[("ids", x) for x in ex_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)
@@ -0,0 +1,50 @@
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()
cve_ids = INPUTS.get("cve_ids", "")
limit = INPUTS.get("limit", "50")
filt = "cve.id:['" + "','".join(csv(cve_ids)) + "']"
result = call("GET", "/spotlight/combined/vulnerabilities/v1", tok, params=[("filter", filt), ("facet", "host_info"), ("limit", limit)])
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)
@@ -0,0 +1,74 @@
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()
ioc_id = INPUTS.get("ioc_id", "")
ind = {"id": ioc_id}
action = INPUTS.get("action", "")
if action:
ind["action"] = action
platforms = INPUTS.get("platforms", "")
if platforms:
ind["platforms"] = csv(platforms)
severity = INPUTS.get("severity", "")
if severity:
ind["severity"] = severity
source = INPUTS.get("source", "")
if source:
ind["source"] = source
description = INPUTS.get("description", "")
if description:
ind["description"] = description
expiration = INPUTS.get("expiration", "")
if expiration:
ind["expiration"] = expiration
result = call("PATCH", "/iocs/entities/indicators/v1", tok, body={"indicators": [ind]})
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)
@@ -0,0 +1,58 @@
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()
host_group_id = INPUTS.get("host_group_id", "")
name = INPUTS.get("name", "")
description = INPUTS.get("description", "")
assignment_rule = INPUTS.get("assignment_rule", "")
res = {"id": host_group_id}
if name:
res["name"] = name
if description:
res["description"] = description
if assignment_rule:
res["assignment_rule"] = assignment_rule
result = call("PATCH", "/devices/entities/host-groups/v1", tok, body={"resources": [res]})
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)
@@ -0,0 +1,77 @@
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()
body = {"id": INPUTS.get("id", "")}
exclusion_name = INPUTS.get("exclusion_name", "")
pattern_id = INPUTS.get("pattern_id", "")
pattern_name = INPUTS.get("pattern_name", "")
cl_regex = INPUTS.get("cl_regex", "")
ifn_regex = INPUTS.get("ifn_regex", "")
comment = INPUTS.get("comment", "")
description = INPUTS.get("description", "")
groups = INPUTS.get("groups", "")
detection_json = INPUTS.get("detection_json", "")
if exclusion_name:
body["name"] = exclusion_name
if pattern_id:
body["pattern_id"] = pattern_id
if pattern_name:
body["pattern_name"] = pattern_name
if cl_regex:
body["cl_regex"] = cl_regex
if ifn_regex:
body["ifn_regex"] = ifn_regex
if comment:
body["comment"] = comment
if description:
body["description"] = description
if groups:
body["groups"] = csv(groups)
if detection_json:
body["detection_json"] = detection_json
result = call("PATCH", "/policy/entities/ioa-exclusions/v1", 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)
@@ -0,0 +1,57 @@
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()
body = {"id": INPUTS.get("id", "")}
value = INPUTS.get("value", "")
comment = INPUTS.get("comment", "")
groups = INPUTS.get("groups", "")
if value:
body["value"] = value
if comment:
body["comment"] = comment
if groups:
body["groups"] = csv(groups)
result = call("PATCH", "/policy/entities/ml-exclusions/v1", 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)
@@ -0,0 +1,76 @@
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()
name = INPUTS.get("name", "")
permission_type = INPUTS.get("permission_type", "private")
content = INPUTS.get("content", "")
boundary = "----RiposteBoundary7a3f91e2"
body_parts = []
for field_name, field_value in [("name", name), ("permission_type", permission_type), ("content", content)]:
body_parts.append(
"--" + boundary + "\r\n"
"Content-Disposition: form-data; name=\"" + field_name + "\"\r\n\r\n"
+ field_value + "\r\n"
)
body_parts.append("--" + boundary + "--\r\n")
body_bytes = "".join(body_parts).encode("utf-8")
headers = {
"Authorization": "Bearer " + tok,
"Accept": "application/json",
"Content-Type": "multipart/form-data; boundary=" + boundary,
}
req = urllib.request.Request(
BASE + "/real-time-response/entities/scripts/v1",
data=body_bytes,
headers=headers,
method="POST",
)
with urllib.request.urlopen(req, timeout=90) as r:
raw = r.read()
result = json.loads(raw) if raw else {}
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)
@@ -0,0 +1,61 @@
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()
definition_id = INPUTS.get("definition_id", "")
name = INPUTS.get("name", "")
key = INPUTS.get("key", "")
source_event_url = INPUTS.get("source_event_url", "")
execution_cid = INPUTS.get("execution_cid", "")
params = [("definition_id", x) for x in csv(definition_id)]
if name:
params.append(("name", name))
params += [("execution_cid", x) for x in csv(execution_cid)]
if key:
params.append(("key", key))
if source_event_url:
params.append(("source_event_url", source_event_url))
json_body = json.loads(INPUTS.get("body") or "{}")
result = call("POST", "/workflows/entities/execute/v1", tok, params=params or None, body=json_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)
@@ -0,0 +1,50 @@
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()
ids = INPUTS.get("ids", "")
action_name = INPUTS.get("action_name", "")
params = [("action_name", action_name)]
result = call("POST", "/workflows/entities/execution-actions/v1", tok, params=params, body={"ids": csv(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)