From 6b8dd13297992feae8b3e1b09e4bd6eaa2bf13d9 Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sun, 12 Jul 2026 00:06:32 +0200 Subject: [PATCH] feat(trend-vision-one): new Trend Vision One XDR integration Vision One API v3.0, 9 commands: list endpoints, isolate/restore endpoint (containment), terminate process, list/get Workbench alerts, add/remove suspicious object (block hash/URL/IP/domain). Bearer-token auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) --- integrations/trend-vision-one/manifest.yaml | 112 ++++++++++++++++++ .../scripts/add_to_block_list.py | 64 ++++++++++ .../trend-vision-one/scripts/get_alert.py | 55 +++++++++ .../scripts/isolate_endpoint.py | 63 ++++++++++ .../trend-vision-one/scripts/list_alerts.py | 57 +++++++++ .../scripts/list_endpoints.py | 55 +++++++++ .../scripts/remove_from_block_list.py | 63 ++++++++++ .../scripts/restore_endpoint.py | 63 ++++++++++ .../scripts/terminate_process.py | 67 +++++++++++ .../scripts/test_connection.py | 50 ++++++++ 10 files changed, 649 insertions(+) create mode 100644 integrations/trend-vision-one/manifest.yaml create mode 100644 integrations/trend-vision-one/scripts/add_to_block_list.py create mode 100644 integrations/trend-vision-one/scripts/get_alert.py create mode 100644 integrations/trend-vision-one/scripts/isolate_endpoint.py create mode 100644 integrations/trend-vision-one/scripts/list_alerts.py create mode 100644 integrations/trend-vision-one/scripts/list_endpoints.py create mode 100644 integrations/trend-vision-one/scripts/remove_from_block_list.py create mode 100644 integrations/trend-vision-one/scripts/restore_endpoint.py create mode 100644 integrations/trend-vision-one/scripts/terminate_process.py create mode 100644 integrations/trend-vision-one/scripts/test_connection.py diff --git a/integrations/trend-vision-one/manifest.yaml b/integrations/trend-vision-one/manifest.yaml new file mode 100644 index 0000000..eeda3cb --- /dev/null +++ b/integrations/trend-vision-one/manifest.yaml @@ -0,0 +1,112 @@ +id: trend_vision_one +name: Trend Vision One +version: 1.0.0 +description: "Trend Micro Vision One (XDR, API v3.0) — endpoint containment and threat response: list endpoints, isolate/restore an endpoint, terminate a process, list and read Workbench alerts, and add/remove suspicious objects (block hash/URL/IP/domain). Bearer-token authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: list endpoints, isolate/restore endpoint, terminate process, list/get Workbench alerts, add/remove suspicious object." +category: endpoint + +# Per-instance configuration. The API token is sent as 'Authorization: Bearer '. +# base_url is the regional Vision One API host. +config_schema: + properties: + base_url: + type: string + description: "Vision One API base URL (e.g. https://api.xdr.trendmicro.com or a regional host)" + default: "https://api.xdr.trendmicro.com" + api_token: + type: string + description: "Vision One API token" + x-soar-sensitive: true + required: + - api_token + +commands: + - id: list_endpoints + name: tmv1-list-endpoints + description: "List endpoints from the endpoint inventory." + risk: read + inputs_schema: + properties: + query: { type: string, description: "Optional filter, e.g. endpointName eq 'host01' (TMV1 query syntax)" } + top: { type: number, description: "Max endpoints (default 50)" } + required: [] + outputs_schema: { properties: {} } + - id: isolate_endpoint + name: tmv1-isolate-endpoint + description: "Isolate an endpoint from the network (containment)." + inputs_schema: + properties: + endpoint_name: { type: string, description: "Endpoint hostname (provide this or agent_guid)" } + agent_guid: { type: string, description: "Agent GUID (provide this or endpoint_name)" } + description: { type: string, description: "Optional reason" } + required: [] + outputs_schema: { properties: {} } + - id: restore_endpoint + name: tmv1-restore-endpoint + description: "Restore (un-isolate) an endpoint's network connection." + inputs_schema: + properties: + endpoint_name: { type: string, description: "Endpoint hostname (provide this or agent_guid)" } + agent_guid: { type: string, description: "Agent GUID (provide this or endpoint_name)" } + description: { type: string, description: "Optional reason" } + required: [] + outputs_schema: { properties: {} } + - id: terminate_process + name: tmv1-terminate-process + description: "Terminate a process on an endpoint by file SHA-1." + inputs_schema: + properties: + endpoint_name: { type: string, description: "Endpoint hostname (provide this or agent_guid)" } + agent_guid: { type: string, description: "Agent GUID (provide this or endpoint_name)" } + file_sha1: { type: string, description: "SHA-1 of the process image to terminate" } + description: { type: string, description: "Optional reason" } + required: [file_sha1] + outputs_schema: { properties: {} } + - id: list_alerts + name: tmv1-list-alerts + description: "List Workbench alerts in a time window." + risk: read + inputs_schema: + properties: + start_datetime: { type: string, description: "ISO-8601 start (e.g. 2024-01-01T00:00:00Z)" } + end_datetime: { type: string, description: "ISO-8601 end" } + top: { type: number, description: "Max alerts (default 50)" } + required: [] + outputs_schema: { properties: {} } + - id: get_alert + name: tmv1-get-alert + description: "Get a single Workbench alert by ID." + risk: read + inputs_schema: + properties: + alert_id: { type: string, description: "Workbench alert ID" } + required: [alert_id] + outputs_schema: { properties: {} } + - id: add_to_block_list + name: tmv1-add-to-block-list + description: "Add a suspicious object (file SHA-1, URL, IP, domain, or sender) to the block list." + inputs_schema: + properties: + object_type: { type: string, description: "One of: fileSha1, url, ip, domain, senderMailAddress" } + object_value: { type: string, description: "The value to block" } + description: { type: string, description: "Optional reason" } + required: [object_type, object_value] + outputs_schema: { properties: {} } + - id: remove_from_block_list + name: tmv1-remove-from-block-list + description: "Remove a suspicious object from the block list." + inputs_schema: + properties: + object_type: { type: string, description: "One of: fileSha1, url, ip, domain, senderMailAddress" } + object_value: { type: string, description: "The value to remove" } + required: [object_type, object_value] + outputs_schema: { properties: {} } + + - id: test_connection + name: tmv1-test-connection + description: "Verify connectivity and the API token (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/trend-vision-one/scripts/add_to_block_list.py b/integrations/trend-vision-one/scripts/add_to_block_list.py new file mode 100644 index 0000000..0cb34b0 --- /dev/null +++ b/integrations/trend-vision-one/scripts/add_to_block_list.py @@ -0,0 +1,64 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("base_url") or "https://api.xdr.trendmicro.com")).rstrip("/") + + +def request(method, path, cfg, body=None, params=None, extra_headers=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + if extra_headers: + headers.update(extra_headers) + 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 _run(fn): + try: + print(json.dumps(fn(_cfg(), _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) + + +_VALID_OBJECT_TYPES = {"fileSha1", "url", "ip", "domain", "senderMailAddress"} + + +def main(cfg, inputs): + object_type = inputs.get("object_type") + object_value = inputs.get("object_value") + description = inputs.get("description") + if not object_type: + raise Exception("object_type is required") + if not object_value: + raise Exception("object_value is required") + if object_type not in _VALID_OBJECT_TYPES: + raise Exception( + "object_type must be one of: " + ", ".join(sorted(_VALID_OBJECT_TYPES)) + ) + entry = {object_type: object_value, "description": description or "Added via Riposte"} + return request("POST", "/v3.0/threatintel/suspiciousObjects", cfg, body=[entry]) + + +_run(main) diff --git a/integrations/trend-vision-one/scripts/get_alert.py b/integrations/trend-vision-one/scripts/get_alert.py new file mode 100644 index 0000000..63f9bc3 --- /dev/null +++ b/integrations/trend-vision-one/scripts/get_alert.py @@ -0,0 +1,55 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("base_url") or "https://api.xdr.trendmicro.com")).rstrip("/") + + +def request(method, path, cfg, body=None, params=None, extra_headers=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + if extra_headers: + headers.update(extra_headers) + 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 _run(fn): + try: + print(json.dumps(fn(_cfg(), _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) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, inputs): + alert_id = inputs.get("alert_id") + if not alert_id: + raise Exception("alert_id is required") + return request("GET", "/v3.0/workbench/alerts/" + q(alert_id), cfg) + + +_run(main) diff --git a/integrations/trend-vision-one/scripts/isolate_endpoint.py b/integrations/trend-vision-one/scripts/isolate_endpoint.py new file mode 100644 index 0000000..dc7f4a1 --- /dev/null +++ b/integrations/trend-vision-one/scripts/isolate_endpoint.py @@ -0,0 +1,63 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("base_url") or "https://api.xdr.trendmicro.com")).rstrip("/") + + +def request(method, path, cfg, body=None, params=None, extra_headers=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + if extra_headers: + headers.update(extra_headers) + 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 _run(fn): + try: + print(json.dumps(fn(_cfg(), _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 _target(inputs): + t = {} + if inputs.get("endpoint_name"): + t["endpointName"] = inputs["endpoint_name"] + elif inputs.get("agent_guid"): + t["agentGuid"] = inputs["agent_guid"] + else: + raise Exception("endpoint_name or agent_guid is required") + if inputs.get("description"): + t["description"] = inputs["description"] + return t + + +def main(cfg, inputs): + target = _target(inputs) + return request("POST", "/v3.0/response/endpoints/isolate", cfg, body=[target]) + + +_run(main) diff --git a/integrations/trend-vision-one/scripts/list_alerts.py b/integrations/trend-vision-one/scripts/list_alerts.py new file mode 100644 index 0000000..9b18337 --- /dev/null +++ b/integrations/trend-vision-one/scripts/list_alerts.py @@ -0,0 +1,57 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("base_url") or "https://api.xdr.trendmicro.com")).rstrip("/") + + +def request(method, path, cfg, body=None, params=None, extra_headers=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + if extra_headers: + headers.update(extra_headers) + 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 _run(fn): + try: + print(json.dumps(fn(_cfg(), _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(cfg, inputs): + start_datetime = inputs.get("start_datetime") + end_datetime = inputs.get("end_datetime") + top = inputs.get("top") + params = { + "startDateTime": start_datetime, + "endDateTime": end_datetime, + "top": int(top or 50), + } + return request("GET", "/v3.0/workbench/alerts", cfg, params=params) + + +_run(main) diff --git a/integrations/trend-vision-one/scripts/list_endpoints.py b/integrations/trend-vision-one/scripts/list_endpoints.py new file mode 100644 index 0000000..04cd9d0 --- /dev/null +++ b/integrations/trend-vision-one/scripts/list_endpoints.py @@ -0,0 +1,55 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("base_url") or "https://api.xdr.trendmicro.com")).rstrip("/") + + +def request(method, path, cfg, body=None, params=None, extra_headers=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + if extra_headers: + headers.update(extra_headers) + 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 _run(fn): + try: + print(json.dumps(fn(_cfg(), _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(cfg, inputs): + query = inputs.get("query") + top = inputs.get("top") + params = {"top": int(top or 50)} + extra_headers = None + if query: + extra_headers = {"TMV1-Filter": query} + return request("GET", "/v3.0/eiqs/endpoints", cfg, params=params, extra_headers=extra_headers) + + +_run(main) diff --git a/integrations/trend-vision-one/scripts/remove_from_block_list.py b/integrations/trend-vision-one/scripts/remove_from_block_list.py new file mode 100644 index 0000000..17c03dd --- /dev/null +++ b/integrations/trend-vision-one/scripts/remove_from_block_list.py @@ -0,0 +1,63 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("base_url") or "https://api.xdr.trendmicro.com")).rstrip("/") + + +def request(method, path, cfg, body=None, params=None, extra_headers=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + if extra_headers: + headers.update(extra_headers) + 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 _run(fn): + try: + print(json.dumps(fn(_cfg(), _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) + + +_VALID_OBJECT_TYPES = {"fileSha1", "url", "ip", "domain", "senderMailAddress"} + + +def main(cfg, inputs): + object_type = inputs.get("object_type") + object_value = inputs.get("object_value") + if not object_type: + raise Exception("object_type is required") + if not object_value: + raise Exception("object_value is required") + if object_type not in _VALID_OBJECT_TYPES: + raise Exception( + "object_type must be one of: " + ", ".join(sorted(_VALID_OBJECT_TYPES)) + ) + entry = {object_type: object_value} + return request("POST", "/v3.0/threatintel/suspiciousObjects/delete", cfg, body=[entry]) + + +_run(main) diff --git a/integrations/trend-vision-one/scripts/restore_endpoint.py b/integrations/trend-vision-one/scripts/restore_endpoint.py new file mode 100644 index 0000000..9310a56 --- /dev/null +++ b/integrations/trend-vision-one/scripts/restore_endpoint.py @@ -0,0 +1,63 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("base_url") or "https://api.xdr.trendmicro.com")).rstrip("/") + + +def request(method, path, cfg, body=None, params=None, extra_headers=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + if extra_headers: + headers.update(extra_headers) + 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 _run(fn): + try: + print(json.dumps(fn(_cfg(), _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 _target(inputs): + t = {} + if inputs.get("endpoint_name"): + t["endpointName"] = inputs["endpoint_name"] + elif inputs.get("agent_guid"): + t["agentGuid"] = inputs["agent_guid"] + else: + raise Exception("endpoint_name or agent_guid is required") + if inputs.get("description"): + t["description"] = inputs["description"] + return t + + +def main(cfg, inputs): + target = _target(inputs) + return request("POST", "/v3.0/response/endpoints/restore", cfg, body=[target]) + + +_run(main) diff --git a/integrations/trend-vision-one/scripts/terminate_process.py b/integrations/trend-vision-one/scripts/terminate_process.py new file mode 100644 index 0000000..9f39835 --- /dev/null +++ b/integrations/trend-vision-one/scripts/terminate_process.py @@ -0,0 +1,67 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("base_url") or "https://api.xdr.trendmicro.com")).rstrip("/") + + +def request(method, path, cfg, body=None, params=None, extra_headers=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + if extra_headers: + headers.update(extra_headers) + 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 _run(fn): + try: + print(json.dumps(fn(_cfg(), _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 _target(inputs): + t = {} + if inputs.get("endpoint_name"): + t["endpointName"] = inputs["endpoint_name"] + elif inputs.get("agent_guid"): + t["agentGuid"] = inputs["agent_guid"] + else: + raise Exception("endpoint_name or agent_guid is required") + if inputs.get("description"): + t["description"] = inputs["description"] + return t + + +def main(cfg, inputs): + file_sha1 = inputs.get("file_sha1") + if not file_sha1: + raise Exception("file_sha1 is required") + target = _target(inputs) + target["fileSha1"] = file_sha1 + return request("POST", "/v3.0/response/endpoints/terminateProcess", cfg, body=[target]) + + +_run(main) diff --git a/integrations/trend-vision-one/scripts/test_connection.py b/integrations/trend-vision-one/scripts/test_connection.py new file mode 100644 index 0000000..cbac3ad --- /dev/null +++ b/integrations/trend-vision-one/scripts/test_connection.py @@ -0,0 +1,50 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("base_url") or "https://api.xdr.trendmicro.com")).rstrip("/") + + +def request(method, path, cfg, body=None, params=None, extra_headers=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + if extra_headers: + headers.update(extra_headers) + 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 _run(fn): + try: + print(json.dumps(fn(_cfg(), _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(cfg, inputs): + request("GET", "/v3.0/eiqs/endpoints", cfg, params={"top": 1}) + return {"ok": True} + + +_run(main)