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) <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -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: {} }
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user