From 1c3936992bafdf671f03bbadec8a5d27f0db21bb Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sun, 12 Jul 2026 00:06:32 +0200 Subject: [PATCH] feat(cybereason): new Cybereason EDR integration Cybereason API, 7 commands: query Malops, list sensors, get machine details, isolate/un-isolate machine (containment), block file hash. Session (login) auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) --- integrations/cybereason/manifest.yaml | 95 +++++++++++++++++++ integrations/cybereason/scripts/block_file.py | 89 +++++++++++++++++ .../cybereason/scripts/get_machine_details.py | 85 +++++++++++++++++ .../cybereason/scripts/isolate_machine.py | 86 +++++++++++++++++ .../cybereason/scripts/list_sensors.py | 90 ++++++++++++++++++ .../cybereason/scripts/query_malops.py | 84 ++++++++++++++++ .../cybereason/scripts/test_connection.py | 76 +++++++++++++++ .../cybereason/scripts/unisolate_machine.py | 86 +++++++++++++++++ 8 files changed, 691 insertions(+) create mode 100644 integrations/cybereason/manifest.yaml create mode 100644 integrations/cybereason/scripts/block_file.py create mode 100644 integrations/cybereason/scripts/get_machine_details.py create mode 100644 integrations/cybereason/scripts/isolate_machine.py create mode 100644 integrations/cybereason/scripts/list_sensors.py create mode 100644 integrations/cybereason/scripts/query_malops.py create mode 100644 integrations/cybereason/scripts/test_connection.py create mode 100644 integrations/cybereason/scripts/unisolate_machine.py diff --git a/integrations/cybereason/manifest.yaml b/integrations/cybereason/manifest.yaml new file mode 100644 index 0000000..c8bbde6 --- /dev/null +++ b/integrations/cybereason/manifest.yaml @@ -0,0 +1,95 @@ +id: cybereason +name: Cybereason +version: 1.0.0 +description: "Cybereason EDR — endpoint containment and investigation: query Malops, list sensors, isolate/un-isolate a machine (containment), block a file hash, and read machine details. Session (login) authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: query Malops, list sensors, isolate/un-isolate machine, block file, get machine details." +category: endpoint + +# Per-instance configuration. Each command logs in (form POST to /login.html, +# reusing the returned session cookie) then performs the action. +config_schema: + properties: + server_url: + type: string + description: "Cybereason server URL (e.g. https://your-tenant.cybereason.net:443)" + username: + type: string + description: "Cybereason username" + password: + type: string + description: "Cybereason password" + x-soar-sensitive: true + insecure: + type: boolean + description: "Trust any TLS certificate (not secure)" + default: false + required: + - server_url + - username + - password + +commands: + - id: query_malops + name: cybereason-query-malops + description: "Query Malops (malicious operations) in a time window." + risk: read + inputs_schema: + properties: + start_time: { type: number, description: "Start time as Unix epoch milliseconds" } + end_time: { type: number, description: "End time as Unix epoch milliseconds" } + required: [] + outputs_schema: { properties: {} } + - id: list_sensors + name: cybereason-list-sensors + description: "List sensors (endpoints)." + risk: read + inputs_schema: + properties: + limit: { type: number, description: "Max sensors (default 100)" } + offset: { type: number, description: "Offset (default 0)" } + filter_field: { type: string, description: "Optional field to filter on (e.g. machineName)" } + filter_value: { type: string, description: "Value for the filter field (equals match)" } + required: [] + outputs_schema: { properties: {} } + - id: get_machine_details + name: cybereason-get-machine-details + description: "Get details for a sensor/machine by its sensor ID." + risk: read + inputs_schema: + properties: + sensor_id: { type: string, description: "Sensor ID" } + required: [sensor_id] + outputs_schema: { properties: {} } + - id: isolate_machine + name: cybereason-isolate-machine + description: "Isolate one or more machines from the network (containment)." + inputs_schema: + properties: + sensor_ids: { type: string, description: "Comma-separated sensor IDs to isolate" } + required: [sensor_ids] + outputs_schema: { properties: {} } + - id: unisolate_machine + name: cybereason-unisolate-machine + description: "Remove one or more machines from isolation." + inputs_schema: + properties: + sensor_ids: { type: string, description: "Comma-separated sensor IDs to un-isolate" } + required: [sensor_ids] + outputs_schema: { properties: {} } + - id: block_file + name: cybereason-block-file + description: "Block a file by hash (add to the block list)." + inputs_schema: + properties: + md5: { type: string, description: "MD5 hash of the file to block" } + required: [md5] + outputs_schema: { properties: {} } + + - id: test_connection + name: cybereason-test-connection + description: "Verify connectivity and credentials by logging in (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/cybereason/scripts/block_file.py b/integrations/cybereason/scripts/block_file.py new file mode 100644 index 0000000..1773afa --- /dev/null +++ b/integrations/cybereason/scripts/block_file.py @@ -0,0 +1,89 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, http.cookiejar, ssl + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _ctx(cfg): + if cfg.get("insecure"): + c = ssl.create_default_context() + c.check_hostname = False + c.verify_mode = ssl.CERT_NONE + return c + return None + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + ctx = _ctx(cfg) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()), + urllib.request.HTTPSHandler(context=ctx) if ctx else urllib.request.HTTPSHandler(), + ) + + def login(self): + form = urllib.parse.urlencode({ + "username": self.cfg.get("username", ""), + "password": self.cfg.get("password", ""), + }).encode("utf-8") + req = urllib.request.Request(self.base + "/login.html", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + r.read() + + def call(self, method, path, body=None): + url = self.base + path + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"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 self.opener.open(req, timeout=90) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"raw": raw.decode("utf-8", "replace")} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.login() + print(json.dumps(fn(client, inputs))) + 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) + + +def main(client, inputs): + md5 = inputs.get("md5") + if not md5: + raise Exception("md5 is required") + + body = [{ + "keys": [{"dataType": "MD5", "value": md5}], + "maliciousType": "blacklist", + "remove": False, + "prventExecution": True, + }] + + resp = client.call("POST", "/rest/classification/update", body) + if not resp: + return {"ok": True, "blocked": md5} + return resp + + +_run(main) diff --git a/integrations/cybereason/scripts/get_machine_details.py b/integrations/cybereason/scripts/get_machine_details.py new file mode 100644 index 0000000..2cf41d7 --- /dev/null +++ b/integrations/cybereason/scripts/get_machine_details.py @@ -0,0 +1,85 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, http.cookiejar, ssl + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _ctx(cfg): + if cfg.get("insecure"): + c = ssl.create_default_context() + c.check_hostname = False + c.verify_mode = ssl.CERT_NONE + return c + return None + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + ctx = _ctx(cfg) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()), + urllib.request.HTTPSHandler(context=ctx) if ctx else urllib.request.HTTPSHandler(), + ) + + def login(self): + form = urllib.parse.urlencode({ + "username": self.cfg.get("username", ""), + "password": self.cfg.get("password", ""), + }).encode("utf-8") + req = urllib.request.Request(self.base + "/login.html", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + r.read() + + def call(self, method, path, body=None): + url = self.base + path + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"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 self.opener.open(req, timeout=90) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"raw": raw.decode("utf-8", "replace")} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.login() + print(json.dumps(fn(client, inputs))) + 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) + + +def main(client, inputs): + sensor_id = inputs.get("sensor_id") + if not sensor_id: + raise Exception("sensor_id is required") + + body = { + "limit": 1, + "offset": 0, + "filters": [{"fieldName": "sensorId", "operator": "Equals", "values": [sensor_id]}], + } + + return client.call("POST", "/rest/sensors/query", body) + + +_run(main) diff --git a/integrations/cybereason/scripts/isolate_machine.py b/integrations/cybereason/scripts/isolate_machine.py new file mode 100644 index 0000000..f6f21d0 --- /dev/null +++ b/integrations/cybereason/scripts/isolate_machine.py @@ -0,0 +1,86 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, http.cookiejar, ssl + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _ctx(cfg): + if cfg.get("insecure"): + c = ssl.create_default_context() + c.check_hostname = False + c.verify_mode = ssl.CERT_NONE + return c + return None + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + ctx = _ctx(cfg) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()), + urllib.request.HTTPSHandler(context=ctx) if ctx else urllib.request.HTTPSHandler(), + ) + + def login(self): + form = urllib.parse.urlencode({ + "username": self.cfg.get("username", ""), + "password": self.cfg.get("password", ""), + }).encode("utf-8") + req = urllib.request.Request(self.base + "/login.html", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + r.read() + + def call(self, method, path, body=None): + url = self.base + path + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"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 self.opener.open(req, timeout=90) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"raw": raw.decode("utf-8", "replace")} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.login() + print(json.dumps(fn(client, inputs))) + 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) + + +def main(client, inputs): + sensor_ids = inputs.get("sensor_ids") + if not sensor_ids: + raise Exception("sensor_ids is required") + + sensor_ids_list = [s.strip() for s in str(sensor_ids).split(",") if s.strip()] + if not sensor_ids_list: + raise Exception("sensor_ids is required") + + resp = client.call("POST", "/rest/sensors/actions/isolate", {"sensorsIds": sensor_ids_list}) + if not resp: + return {"ok": True, "isolated": sensor_ids_list} + return resp + + +_run(main) diff --git a/integrations/cybereason/scripts/list_sensors.py b/integrations/cybereason/scripts/list_sensors.py new file mode 100644 index 0000000..5b78a98 --- /dev/null +++ b/integrations/cybereason/scripts/list_sensors.py @@ -0,0 +1,90 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, http.cookiejar, ssl + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _ctx(cfg): + if cfg.get("insecure"): + c = ssl.create_default_context() + c.check_hostname = False + c.verify_mode = ssl.CERT_NONE + return c + return None + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + ctx = _ctx(cfg) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()), + urllib.request.HTTPSHandler(context=ctx) if ctx else urllib.request.HTTPSHandler(), + ) + + def login(self): + form = urllib.parse.urlencode({ + "username": self.cfg.get("username", ""), + "password": self.cfg.get("password", ""), + }).encode("utf-8") + req = urllib.request.Request(self.base + "/login.html", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + r.read() + + def call(self, method, path, body=None): + url = self.base + path + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"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 self.opener.open(req, timeout=90) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"raw": raw.decode("utf-8", "replace")} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.login() + print(json.dumps(fn(client, inputs))) + 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) + + +def main(client, inputs): + limit = inputs.get("limit") + offset = inputs.get("offset") + filter_field = inputs.get("filter_field") + filter_value = inputs.get("filter_value") + + filters = [] + if filter_field not in (None, "") and filter_value not in (None, ""): + filters = [{"fieldName": filter_field, "operator": "Equals", "values": [filter_value]}] + + body = { + "limit": int(limit) if limit not in (None, "") else 100, + "offset": int(offset) if offset not in (None, "") else 0, + "filters": filters, + } + + return client.call("POST", "/rest/sensors/query", body) + + +_run(main) diff --git a/integrations/cybereason/scripts/query_malops.py b/integrations/cybereason/scripts/query_malops.py new file mode 100644 index 0000000..779c049 --- /dev/null +++ b/integrations/cybereason/scripts/query_malops.py @@ -0,0 +1,84 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, http.cookiejar, ssl + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _ctx(cfg): + if cfg.get("insecure"): + c = ssl.create_default_context() + c.check_hostname = False + c.verify_mode = ssl.CERT_NONE + return c + return None + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + ctx = _ctx(cfg) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()), + urllib.request.HTTPSHandler(context=ctx) if ctx else urllib.request.HTTPSHandler(), + ) + + def login(self): + form = urllib.parse.urlencode({ + "username": self.cfg.get("username", ""), + "password": self.cfg.get("password", ""), + }).encode("utf-8") + req = urllib.request.Request(self.base + "/login.html", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + r.read() + + def call(self, method, path, body=None): + url = self.base + path + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"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 self.opener.open(req, timeout=90) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"raw": raw.decode("utf-8", "replace")} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.login() + print(json.dumps(fn(client, inputs))) + 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) + + +def main(client, inputs): + start_time = inputs.get("start_time") + end_time = inputs.get("end_time") + + body = {} + if start_time not in (None, ""): + body["startTime"] = int(start_time) + if end_time not in (None, ""): + body["endTime"] = int(end_time) + + return client.call("POST", "/rest/detection/inbox", body) + + +_run(main) diff --git a/integrations/cybereason/scripts/test_connection.py b/integrations/cybereason/scripts/test_connection.py new file mode 100644 index 0000000..18eea62 --- /dev/null +++ b/integrations/cybereason/scripts/test_connection.py @@ -0,0 +1,76 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, http.cookiejar, ssl + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _ctx(cfg): + if cfg.get("insecure"): + c = ssl.create_default_context() + c.check_hostname = False + c.verify_mode = ssl.CERT_NONE + return c + return None + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + ctx = _ctx(cfg) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()), + urllib.request.HTTPSHandler(context=ctx) if ctx else urllib.request.HTTPSHandler(), + ) + + def login(self): + form = urllib.parse.urlencode({ + "username": self.cfg.get("username", ""), + "password": self.cfg.get("password", ""), + }).encode("utf-8") + req = urllib.request.Request(self.base + "/login.html", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + r.read() + + def call(self, method, path, body=None): + url = self.base + path + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"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 self.opener.open(req, timeout=90) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"raw": raw.decode("utf-8", "replace")} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.login() + print(json.dumps(fn(client, inputs))) + 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) + + +def main(client, inputs): + client.call("POST", "/rest/sensors/query", {"limit": 1, "offset": 0, "filters": []}) + return {"ok": True} + + +_run(main) diff --git a/integrations/cybereason/scripts/unisolate_machine.py b/integrations/cybereason/scripts/unisolate_machine.py new file mode 100644 index 0000000..2a2709f --- /dev/null +++ b/integrations/cybereason/scripts/unisolate_machine.py @@ -0,0 +1,86 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, http.cookiejar, ssl + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _ctx(cfg): + if cfg.get("insecure"): + c = ssl.create_default_context() + c.check_hostname = False + c.verify_mode = ssl.CERT_NONE + return c + return None + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + ctx = _ctx(cfg) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()), + urllib.request.HTTPSHandler(context=ctx) if ctx else urllib.request.HTTPSHandler(), + ) + + def login(self): + form = urllib.parse.urlencode({ + "username": self.cfg.get("username", ""), + "password": self.cfg.get("password", ""), + }).encode("utf-8") + req = urllib.request.Request(self.base + "/login.html", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + r.read() + + def call(self, method, path, body=None): + url = self.base + path + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"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 self.opener.open(req, timeout=90) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"raw": raw.decode("utf-8", "replace")} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.login() + print(json.dumps(fn(client, inputs))) + 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) + + +def main(client, inputs): + sensor_ids = inputs.get("sensor_ids") + if not sensor_ids: + raise Exception("sensor_ids is required") + + sensor_ids_list = [s.strip() for s in str(sensor_ids).split(",") if s.strip()] + if not sensor_ids_list: + raise Exception("sensor_ids is required") + + resp = client.call("POST", "/rest/sensors/actions/unisolate", {"sensorsIds": sensor_ids_list}) + if not resp: + return {"ok": True, "unisolated": sensor_ids_list} + return resp + + +_run(main)