From ef215daa88d5883eee139f6451af49ec070cb39d Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sat, 27 Jun 2026 15:04:22 +0200 Subject: [PATCH] feat(sekoia): Sekoia XDR + SEKOIA Intelligence Center integrations Sekoia XDR (siem): alert ingestion (list_alerts) with an exhaustive OCSF mapper and a bundled 'Sekoia XDR Alert' default type, plus 20 commands across alerts (list/get/search, status workflow, comments), event search jobs (create/status/ results + one-shot search_events), cases, asset management, users, kill chains and a generic authenticated HTTP passthrough. Bearer-token auth, EU host default. SEKOIA Intelligence Center (enrichment): observable/indicator/indicator-context CTI queries plus ip/url/domain/file/email reputation lookups (STIX type resolved automatically). No fetch. Co-Authored-By: Claude Opus 4.8 --- .../sekoia-intelligence-center/manifest.yaml | 122 ++++++++ .../scripts/domain.py | 37 +++ .../scripts/email.py | 37 +++ .../scripts/file.py | 37 +++ .../scripts/get_indicator.py | 41 +++ .../scripts/get_indicator_context.py | 41 +++ .../scripts/get_observable.py | 41 +++ .../sekoia-intelligence-center/scripts/ip.py | 38 +++ .../scripts/test_connection.py | 33 +++ .../sekoia-intelligence-center/scripts/url.py | 37 +++ .../incident-types/sekoia-xdr-alert.yaml | 3 + integrations/sekoia-xdr/manifest.yaml | 279 ++++++++++++++++++ .../sekoia-xdr/mappers/list_alerts.yaml | 43 +++ .../scripts/add_attributes_asset.py | 42 +++ .../sekoia-xdr/scripts/add_keys_asset.py | 42 +++ .../scripts/events_execute_query.py | 39 +++ .../scripts/events_results_query.py | 35 +++ .../sekoia-xdr/scripts/events_status_query.py | 35 +++ integrations/sekoia-xdr/scripts/get_alert.py | 35 +++ integrations/sekoia-xdr/scripts/get_asset.py | 35 +++ .../sekoia-xdr/scripts/get_cases_alert.py | 43 +++ .../sekoia-xdr/scripts/get_comments.py | 35 +++ .../sekoia-xdr/scripts/get_kill_chain.py | 35 +++ integrations/sekoia-xdr/scripts/get_user.py | 35 +++ .../sekoia-xdr/scripts/get_workflow_alert.py | 35 +++ .../sekoia-xdr/scripts/http_request.py | 56 ++++ .../sekoia-xdr/scripts/list_alerts.py | 63 ++++ .../sekoia-xdr/scripts/list_assets.py | 43 +++ .../sekoia-xdr/scripts/post_comment_alert.py | 40 +++ .../scripts/remove_attribute_asset.py | 37 +++ .../sekoia-xdr/scripts/remove_key_asset.py | 37 +++ .../sekoia-xdr/scripts/search_events.py | 50 ++++ .../sekoia-xdr/scripts/test_connection.py | 33 +++ .../sekoia-xdr/scripts/update_status_alert.py | 52 ++++ 34 files changed, 1646 insertions(+) create mode 100644 integrations/sekoia-intelligence-center/manifest.yaml create mode 100644 integrations/sekoia-intelligence-center/scripts/domain.py create mode 100644 integrations/sekoia-intelligence-center/scripts/email.py create mode 100644 integrations/sekoia-intelligence-center/scripts/file.py create mode 100644 integrations/sekoia-intelligence-center/scripts/get_indicator.py create mode 100644 integrations/sekoia-intelligence-center/scripts/get_indicator_context.py create mode 100644 integrations/sekoia-intelligence-center/scripts/get_observable.py create mode 100644 integrations/sekoia-intelligence-center/scripts/ip.py create mode 100644 integrations/sekoia-intelligence-center/scripts/test_connection.py create mode 100644 integrations/sekoia-intelligence-center/scripts/url.py create mode 100644 integrations/sekoia-xdr/incident-types/sekoia-xdr-alert.yaml create mode 100644 integrations/sekoia-xdr/manifest.yaml create mode 100644 integrations/sekoia-xdr/mappers/list_alerts.yaml create mode 100644 integrations/sekoia-xdr/scripts/add_attributes_asset.py create mode 100644 integrations/sekoia-xdr/scripts/add_keys_asset.py create mode 100644 integrations/sekoia-xdr/scripts/events_execute_query.py create mode 100644 integrations/sekoia-xdr/scripts/events_results_query.py create mode 100644 integrations/sekoia-xdr/scripts/events_status_query.py create mode 100644 integrations/sekoia-xdr/scripts/get_alert.py create mode 100644 integrations/sekoia-xdr/scripts/get_asset.py create mode 100644 integrations/sekoia-xdr/scripts/get_cases_alert.py create mode 100644 integrations/sekoia-xdr/scripts/get_comments.py create mode 100644 integrations/sekoia-xdr/scripts/get_kill_chain.py create mode 100644 integrations/sekoia-xdr/scripts/get_user.py create mode 100644 integrations/sekoia-xdr/scripts/get_workflow_alert.py create mode 100644 integrations/sekoia-xdr/scripts/http_request.py create mode 100644 integrations/sekoia-xdr/scripts/list_alerts.py create mode 100644 integrations/sekoia-xdr/scripts/list_assets.py create mode 100644 integrations/sekoia-xdr/scripts/post_comment_alert.py create mode 100644 integrations/sekoia-xdr/scripts/remove_attribute_asset.py create mode 100644 integrations/sekoia-xdr/scripts/remove_key_asset.py create mode 100644 integrations/sekoia-xdr/scripts/search_events.py create mode 100644 integrations/sekoia-xdr/scripts/test_connection.py create mode 100644 integrations/sekoia-xdr/scripts/update_status_alert.py diff --git a/integrations/sekoia-intelligence-center/manifest.yaml b/integrations/sekoia-intelligence-center/manifest.yaml new file mode 100644 index 0000000..079ba6d --- /dev/null +++ b/integrations/sekoia-intelligence-center/manifest.yaml @@ -0,0 +1,122 @@ +id: sekoia_intelligence_center +name: SEKOIA Intelligence Center +version: 1.0.0 +description: "SEKOIA.IO Intelligence Center (CTI) — query observables, indicators and full indicator context (STIX bundles), plus reputation lookups for IPs, URLs, domains, files and emails." +changelog: "1.0.0 — Initial release: observable/indicator/indicator-context queries and ip/url/domain/file/email reputation lookups." +category: enrichment + +# Per-instance configuration. An Organization API key is sent as a Bearer token. +config_schema: + properties: + url: + type: string + description: "Sekoia API base URL, e.g. https://api.sekoia.io" + default: https://api.sekoia.io + api_key: + type: string + description: "Sekoia API key (Bearer token)" + x-soar-sensitive: true + required: + - url + - api_key + +auth: + - id: bearer + type: api_key + in: header + name: Authorization + value_template: "Bearer {{secret}}" + secret_field: api_key + +commands: + - id: get_observable + name: sekoia-intelligence-get-observable + description: "Query the Intelligence Center for information about an observable (value + type)." + risk: read + inputs_schema: + properties: + value: { type: string, description: "Observable value" } + type: { type: string, description: "Observable type (e.g. ipv4-addr, domain-name, url, file, email-addr)" } + required: [value, type] + outputs_schema: { properties: {} } + + - id: get_indicator + name: sekoia-intelligence-get-indicator + description: "Query the Intelligence Center for information about an indicator (IoC) by value + type. Returns nothing if the value is not a known indicator." + risk: read + inputs_schema: + properties: + value: { type: string, description: "Indicator value" } + type: { type: string, description: "Indicator type (e.g. ipv4-addr, domain-name, url, file, email-addr)" } + required: [value, type] + outputs_schema: { properties: {} } + + - id: get_indicator_context + name: sekoia-intelligence-get-indicator-context + description: "Retrieve the full context (STIX bundle: related malware, threat actors, kill chain, etc.) for an indicator by value + type." + risk: read + inputs_schema: + properties: + value: { type: string, description: "Indicator value" } + type: { type: string, description: "Indicator type (e.g. ipv4-addr, domain-name, url, file, email-addr)" } + required: [value, type] + outputs_schema: { properties: {} } + + - id: ip + name: sekoia-intelligence-ip + description: "Reputation lookup for an IP address (resolves the STIX type automatically, IPv4 or IPv6)." + risk: read + inputs_schema: + properties: + ip: { type: string, description: "IP address" } + required: [ip] + outputs_schema: { properties: {} } + + - id: url + name: sekoia-intelligence-url + description: "Reputation lookup for a URL." + risk: read + inputs_schema: + properties: + url: { type: string, description: "URL" } + required: [url] + outputs_schema: { properties: {} } + + - id: domain + name: sekoia-intelligence-domain + description: "Reputation lookup for a domain name." + risk: read + inputs_schema: + properties: + domain: { type: string, description: "Domain name" } + required: [domain] + outputs_schema: { properties: {} } + + - id: file + name: sekoia-intelligence-file + description: "Reputation lookup for a file hash." + risk: read + inputs_schema: + properties: + file: { type: string, description: "File hash (MD5, SHA1 or SHA256)" } + required: [file] + outputs_schema: { properties: {} } + + - id: email + name: sekoia-intelligence-email + description: "Reputation lookup for an email address." + risk: read + inputs_schema: + properties: + email: { type: string, description: "Email address" } + required: [email] + outputs_schema: { properties: {} } + + - id: test_connection + name: sekoia-intelligence-test-connection + description: "Verify connectivity and credentials (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/sekoia-intelligence-center/scripts/domain.py b/integrations/sekoia-intelligence-center/scripts/domain.py new file mode 100644 index 0000000..ed47b29 --- /dev/null +++ b/integrations/sekoia-intelligence-center/scripts/domain.py @@ -0,0 +1,37 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, params): + base, headers = _cfg() + clean = {k: v for k, v in params.items() if v not in (None, "")} + full = base + "/" + path.lstrip("/") + ("?" + urllib.parse.urlencode(clean, doseq=True) if clean else "") + req = urllib.request.Request(full, headers=headers, method="GET") + with urllib.request.urlopen(req, timeout=90) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def run(): + value = str(_inputs().get("domain", "")) + print(json.dumps(request("/v2/inthreat/indicators/context", {"value": value, "type": "domain-name"}))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-intelligence-center/scripts/email.py b/integrations/sekoia-intelligence-center/scripts/email.py new file mode 100644 index 0000000..b0a55b0 --- /dev/null +++ b/integrations/sekoia-intelligence-center/scripts/email.py @@ -0,0 +1,37 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, params): + base, headers = _cfg() + clean = {k: v for k, v in params.items() if v not in (None, "")} + full = base + "/" + path.lstrip("/") + ("?" + urllib.parse.urlencode(clean, doseq=True) if clean else "") + req = urllib.request.Request(full, headers=headers, method="GET") + with urllib.request.urlopen(req, timeout=90) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def run(): + value = str(_inputs().get("email", "")) + print(json.dumps(request("/v2/inthreat/indicators/context", {"value": value, "type": "email-addr"}))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-intelligence-center/scripts/file.py b/integrations/sekoia-intelligence-center/scripts/file.py new file mode 100644 index 0000000..6712c7c --- /dev/null +++ b/integrations/sekoia-intelligence-center/scripts/file.py @@ -0,0 +1,37 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, params): + base, headers = _cfg() + clean = {k: v for k, v in params.items() if v not in (None, "")} + full = base + "/" + path.lstrip("/") + ("?" + urllib.parse.urlencode(clean, doseq=True) if clean else "") + req = urllib.request.Request(full, headers=headers, method="GET") + with urllib.request.urlopen(req, timeout=90) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def run(): + value = str(_inputs().get("file", "")) + print(json.dumps(request("/v2/inthreat/indicators/context", {"value": value, "type": "file"}))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-intelligence-center/scripts/get_indicator.py b/integrations/sekoia-intelligence-center/scripts/get_indicator.py new file mode 100644 index 0000000..e9af81c --- /dev/null +++ b/integrations/sekoia-intelligence-center/scripts/get_indicator.py @@ -0,0 +1,41 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path, params=None): + base, headers = _cfg() + url = base + "/" + path.lstrip("/") + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean, doseq=True) + req = urllib.request.Request(url, 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(): + inp = _inputs() + params = {"value": inp.get("value"), "type": inp.get("type")} + print(json.dumps(request("GET", "/v2/inthreat/indicators", params=params))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-intelligence-center/scripts/get_indicator_context.py b/integrations/sekoia-intelligence-center/scripts/get_indicator_context.py new file mode 100644 index 0000000..d2d2e82 --- /dev/null +++ b/integrations/sekoia-intelligence-center/scripts/get_indicator_context.py @@ -0,0 +1,41 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path, params=None): + base, headers = _cfg() + url = base + "/" + path.lstrip("/") + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean, doseq=True) + req = urllib.request.Request(url, 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(): + inp = _inputs() + params = {"value": inp.get("value"), "type": inp.get("type")} + print(json.dumps(request("GET", "/v2/inthreat/indicators/context", params=params))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-intelligence-center/scripts/get_observable.py b/integrations/sekoia-intelligence-center/scripts/get_observable.py new file mode 100644 index 0000000..043ee21 --- /dev/null +++ b/integrations/sekoia-intelligence-center/scripts/get_observable.py @@ -0,0 +1,41 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path, params=None): + base, headers = _cfg() + url = base + "/" + path.lstrip("/") + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean, doseq=True) + req = urllib.request.Request(url, 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(): + inp = _inputs() + params = {"match[value]": inp.get("value"), "match[type]": inp.get("type")} + print(json.dumps(request("GET", "/v2/inthreat/observables", params=params))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-intelligence-center/scripts/ip.py b/integrations/sekoia-intelligence-center/scripts/ip.py new file mode 100644 index 0000000..6115e17 --- /dev/null +++ b/integrations/sekoia-intelligence-center/scripts/ip.py @@ -0,0 +1,38 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, params): + base, headers = _cfg() + clean = {k: v for k, v in params.items() if v not in (None, "")} + url = base + "/" + path.lstrip("/") + ("?" + urllib.parse.urlencode(clean, doseq=True) if clean else "") + req = urllib.request.Request(url, headers=headers, method="GET") + with urllib.request.urlopen(req, timeout=90) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def run(): + value = str(_inputs().get("ip", "")) + stix_type = "ipv6-addr" if ":" in value else "ipv4-addr" + print(json.dumps(request("/v2/inthreat/indicators/context", {"value": value, "type": stix_type}))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-intelligence-center/scripts/test_connection.py b/integrations/sekoia-intelligence-center/scripts/test_connection.py new file mode 100644 index 0000000..8e2e740 --- /dev/null +++ b/integrations/sekoia-intelligence-center/scripts/test_connection.py @@ -0,0 +1,33 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def request(method, path): + base, headers = _cfg() + req = urllib.request.Request(base + "/" + path.lstrip("/"), headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def run(): + request("GET", "/v1/auth/validate") + print(json.dumps({"ok": True})) + + +try: + run() +except urllib.error.HTTPError as e: + detail = e.read().decode("utf-8", "replace") + msg = "API key is not valid." if e.code in (401, 403) else "HTTP " + str(e.code) + print(json.dumps({"ok": False, "error": msg, "detail": detail})) + sys.exit(1) +except Exception as e: + print(json.dumps({"ok": False, "error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-intelligence-center/scripts/url.py b/integrations/sekoia-intelligence-center/scripts/url.py new file mode 100644 index 0000000..719d08b --- /dev/null +++ b/integrations/sekoia-intelligence-center/scripts/url.py @@ -0,0 +1,37 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, params): + base, headers = _cfg() + clean = {k: v for k, v in params.items() if v not in (None, "")} + full = base + "/" + path.lstrip("/") + ("?" + urllib.parse.urlencode(clean, doseq=True) if clean else "") + req = urllib.request.Request(full, headers=headers, method="GET") + with urllib.request.urlopen(req, timeout=90) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def run(): + value = str(_inputs().get("url", "")) + print(json.dumps(request("/v2/inthreat/indicators/context", {"value": value, "type": "url"}))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/incident-types/sekoia-xdr-alert.yaml b/integrations/sekoia-xdr/incident-types/sekoia-xdr-alert.yaml new file mode 100644 index 0000000..b1bebe4 --- /dev/null +++ b/integrations/sekoia-xdr/incident-types/sekoia-xdr-alert.yaml @@ -0,0 +1,3 @@ +name: "Sekoia XDR Alert" +color: "#6c5ce7" +icon: "alert" diff --git a/integrations/sekoia-xdr/manifest.yaml b/integrations/sekoia-xdr/manifest.yaml new file mode 100644 index 0000000..37055f1 --- /dev/null +++ b/integrations/sekoia-xdr/manifest.yaml @@ -0,0 +1,279 @@ +id: sekoia_xdr +name: Sekoia XDR +version: 1.0.0 +description: "Sekoia Defend (XDR) REST API — alert ingestion and lifecycle (list/get/search, status workflow, comments), event search jobs (create/status/results + one-shot search), cases, asset management (get/list, attributes/keys add/remove), users, kill chains and a generic authenticated HTTP passthrough." +changelog: "1.0.0 — Initial release: alert ingestion (list_alerts) with an exhaustive OCSF mapper, 20 commands across alerts, event search, cases, assets, users and kill chains." +category: siem + +# Per-instance configuration. The API host defaults to the EU region; an +# Organization API key is sent as a Bearer token. +config_schema: + properties: + url: + type: string + description: "Sekoia API base URL, e.g. https://api.sekoia.io" + default: https://api.sekoia.io + api_key: + type: string + description: "Sekoia API key (Bearer token)" + x-soar-sensitive: true + required: + - url + - api_key + +auth: + - id: bearer + type: api_key + in: header + name: Authorization + value_template: "Bearer {{secret}}" + secret_field: api_key + +commands: + # ── Ingestion ─────────────────────────────────────────────────────────────── + - id: list_alerts + name: sekoia-xdr-list-alerts + description: "List alerts, sorted by created_at. Used for ingestion: results path = items. Alerts aggregate the matching events and are tied to rules and a kill chain." + risk: read + inputs_schema: + properties: + limit: { type: number, description: "Number of alerts to return (1-100, default 20)" } + status: { type: string, description: "Comma-separated statuses to include (Pending, Acknowledged, Ongoing, Rejected, Closed)" } + created_at: { type: string, description: "Filter by creation date. A bare date is treated as the lower bound (',now'); a range ',' is passed through. Incremental fetch watermark." } + updated_at: { type: string, description: "Filter by update date, same format as created_at" } + urgency: { type: string, description: "Urgency range 'MIN,MAX' (e.g. 80,100)" } + alerts_type: { type: string, description: "Comma-separated alert categories (e.g. malware,phishing)" } + sort_by: { type: string, description: "Field to sort by (created_at, updated_at, target, urgency, status). Default created_at." } + direction: { type: string, description: "Sort direction (asc, desc). Default asc." } + offset: { type: number, description: "Result offset for paging" } + required: [] + outputs_schema: { properties: {} } + ingest: + results_path: items + dedup_key: uuid + incremental_field: created_at + + - id: get_alert + name: sekoia-xdr-get-alert + description: "Retrieve a single alert by UUID or short_id." + risk: read + inputs_schema: + properties: + id: { type: string, description: "Alert UUID or short_id (e.g. ALUnyZCYZ9Ga)" } + required: [id] + outputs_schema: { properties: {} } + + - id: search_events + name: sekoia-xdr-search-events + description: "Search events for an alert in one call: creates an event search job, waits for it to finish and returns the events. Falls back to returning the job UUID if it is still running." + risk: read + inputs_schema: + properties: + earliest_time: { type: string, description: "Lower time bound (e.g. -3d, -2w, or ISO 8601 2023-01-15T00:00:00Z)" } + latest_time: { type: string, description: "Upper time bound (e.g. now, +1d, or ISO 8601)" } + query: { type: string, description: "Event query, e.g. alert_short_ids:ALUnyZCYZ9Ga" } + max_last_events: { type: string, description: "Maximum number of events to return" } + required: [earliest_time, latest_time] + outputs_schema: { properties: {} } + + - id: events_execute_query + name: sekoia-xdr-events-execute-query + description: "Create an event search job. Use events_status_query then events_results_query to poll and retrieve, or use search_events for a one-shot." + risk: read + inputs_schema: + properties: + earliest_time: { type: string, description: "Lower time bound (e.g. -3d or ISO 8601)" } + latest_time: { type: string, description: "Upper time bound (e.g. now or ISO 8601)" } + query: { type: string, description: "Event query, e.g. alert_short_ids:ALUnyZCYZ9Ga" } + max_last_events: { type: string, description: "Maximum number of events to return" } + required: [earliest_time, latest_time] + outputs_schema: { properties: {} } + + - id: events_status_query + name: sekoia-xdr-events-status-query + description: "Get the status of a previously created event search job." + risk: read + inputs_schema: + properties: + uuid: { type: string, description: "Event search job UUID" } + required: [uuid] + outputs_schema: { properties: {} } + + - id: events_results_query + name: sekoia-xdr-events-results-query + description: "Retrieve the events of a finished event search job." + risk: read + inputs_schema: + properties: + uuid: { type: string, description: "Event search job UUID" } + required: [uuid] + outputs_schema: { properties: {} } + + - id: update_status_alert + name: sekoia-xdr-update-status-alert + description: "Change the status of an alert by applying the matching workflow transition (Acknowledged, Rejected, Ongoing, Closed)." + risk: safe_write + inputs_schema: + properties: + id: { type: string, description: "Alert UUID or short_id" } + status: { type: string, description: "New status: Acknowledged, Rejected, Ongoing or Closed" } + comment: { type: string, description: "Optional comment describing the status change" } + required: [id, status] + outputs_schema: { properties: {} } + + - id: post_comment_alert + name: sekoia-xdr-post-comment-alert + description: "Post a comment on an alert." + risk: safe_write + inputs_schema: + properties: + id: { type: string, description: "Alert UUID or short_id" } + comment: { type: string, description: "Comment content" } + author: { type: string, description: "Optional comment author" } + required: [id, comment] + outputs_schema: { properties: {} } + + - id: get_comments + name: sekoia-xdr-get-comments + description: "Get all comments of an alert." + risk: read + inputs_schema: + properties: + id: { type: string, description: "Alert UUID or short_id" } + required: [id] + outputs_schema: { properties: {} } + + - id: get_workflow_alert + name: sekoia-xdr-get-workflow-alert + description: "Get the available status transitions for an alert." + risk: read + inputs_schema: + properties: + id: { type: string, description: "Alert UUID or short_id" } + required: [id] + outputs_schema: { properties: {} } + + - id: get_cases_alert + name: sekoia-xdr-get-cases-alert + description: "Retrieve the cases related to an alert. If case_id is given, only that case is returned." + risk: read + inputs_schema: + properties: + alert_id: { type: string, description: "Alert UUID or short_id" } + case_id: { type: string, description: "Optional case short_id (e.g. CAQNurTJM8q2)" } + required: [alert_id] + outputs_schema: { properties: {} } + + - id: get_asset + name: sekoia-xdr-get-asset + description: "Get an asset by its UUID." + risk: read + inputs_schema: + properties: + asset_uuid: { type: string, description: "Asset UUID" } + required: [asset_uuid] + outputs_schema: { properties: {} } + + - id: list_assets + name: sekoia-xdr-list-assets + description: "List assets, optionally filtered by type." + risk: read + inputs_schema: + properties: + limit: { type: number, description: "Number of assets to return (default 10)" } + assets_type: { type: string, description: "Filter by asset type name (e.g. computer, network)" } + required: [] + outputs_schema: { properties: {} } + + - id: get_user + name: sekoia-xdr-get-user + description: "Get information about a user by UUID (e.g. a comment author)." + risk: read + inputs_schema: + properties: + user_uuid: { type: string, description: "User UUID" } + required: [user_uuid] + outputs_schema: { properties: {} } + + - id: add_attributes_asset + name: sekoia-xdr-add-attributes-asset + description: "Add an attribute (name/value) to an asset." + risk: safe_write + inputs_schema: + properties: + asset_uuid: { type: string, description: "Asset UUID" } + name: { type: string, description: "Attribute name" } + value: { type: string, description: "Attribute value" } + required: [asset_uuid, name, value] + outputs_schema: { properties: {} } + + - id: add_keys_asset + name: sekoia-xdr-add-keys-asset + description: "Add a key (name/value) to an asset." + risk: safe_write + inputs_schema: + properties: + asset_uuid: { type: string, description: "Asset UUID" } + name: { type: string, description: "Key name" } + value: { type: string, description: "Key value" } + required: [asset_uuid, name, value] + outputs_schema: { properties: {} } + + - id: remove_attribute_asset + name: sekoia-xdr-remove-attribute-asset + description: "Remove an attribute from an asset. Use get_asset to find the attribute UUID." + risk: destructive + inputs_schema: + properties: + asset_uuid: { type: string, description: "Asset UUID" } + attribute_uuid: { type: string, description: "UUID of the attribute to remove" } + required: [asset_uuid, attribute_uuid] + outputs_schema: { properties: {} } + + - id: remove_key_asset + name: sekoia-xdr-remove-key-asset + description: "Remove a key from an asset. Use get_asset to find the key UUID." + risk: destructive + inputs_schema: + properties: + asset_uuid: { type: string, description: "Asset UUID" } + key_uuid: { type: string, description: "UUID of the key to remove" } + required: [asset_uuid, key_uuid] + outputs_schema: { properties: {} } + + - id: get_kill_chain + name: sekoia-xdr-get-kill-chain + description: "Retrieve the definition of a Cyber Kill Chain step by UUID or short_id." + risk: read + inputs_schema: + properties: + kill_chain_uuid: { type: string, description: "Kill chain UUID or short_id" } + required: [kill_chain_uuid] + outputs_schema: { properties: {} } + + - id: http_request + name: sekoia-xdr-http-request + description: "Perform an authenticated HTTP request to the Sekoia API. Advanced escape hatch for endpoints without a dedicated command." + inputs_schema: + properties: + method: { type: string, description: "HTTP method (GET, POST, PUT, PATCH, DELETE). Default GET." } + url_suffix: { type: string, description: "Path after the base URL, e.g. /v1/sic/alerts/ or /v1/asset-management/assets/" } + parameters: { type: string, description: "Query parameters as a JSON object string, e.g. {\"limit\": 10}" } + data: { type: string, description: "Request body as a JSON object string (for POST/PUT/PATCH)" } + required: [url_suffix] + outputs_schema: { properties: {} } + + # ── Connectivity test ───────────────────────────────────────────────────── + - id: test_connection + name: sekoia-xdr-test-connection + description: "Verify connectivity and credentials (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } + +ingestion: + command: list_alerts + mapper: list_alerts + default_incident_type: "Sekoia XDR Alert" diff --git a/integrations/sekoia-xdr/mappers/list_alerts.yaml b/integrations/sekoia-xdr/mappers/list_alerts.yaml new file mode 100644 index 0000000..3d2fd45 --- /dev/null +++ b/integrations/sekoia-xdr/mappers/list_alerts.yaml @@ -0,0 +1,43 @@ +name: "Sekoia XDR Alerts → OCSF" +description: "Maps a Sekoia Defend (XDR) alert (/v1/sic/alerts, results_path = items) to OCSF Detection Finding fields. Alerts aggregate the matching events; use search_events for per-event detail." +field_mappings: + title: "title" + description: "details" + # urgency is a 0-100 score; bucket it onto the 1-5 incident scale. No parentheses (the engine chains + # right-associatively) and the lowest band uses '>= 0 ? 1' so the literal 1 sits in a then-branch — + # a bare trailing else is read as a field path, not an integer, and would fall back to the default. + severity: "urgency.value >= 80 ? 5 : urgency.value >= 60 ? 4 : urgency.value >= 40 ? 3 : urgency.value >= 20 ? 2 : urgency.value >= 0 ? 1 : 1" + # How the alert was generated. + source: "source" +# results_path = items; source_path is JSONata over ONE alert object. +# Paths absent from a given alert return nothing and are skipped. +ocsf: + # ── Finding ─────────────────────────────────────────────────────── + - { source_path: "uuid", ocsf_field: "finding_info.uid" } + - { source_path: "title", ocsf_field: "finding_info.title" } + - { source_path: "rule.description", ocsf_field: "finding_info.desc" } + - { source_path: "created_at", ocsf_field: "finding_info.created_time" } + - { source_path: "updated_at", ocsf_field: "finding_info.modified_time" } + - { source_path: "first_seen_at", ocsf_field: "finding_info.first_seen_time" } + - { source_path: "last_seen_at", ocsf_field: "finding_info.last_seen_time" } + # ── Detection rule (analytic) ───────────────────────────────────── + - { source_path: "rule.name", ocsf_field: "finding_info.analytic.name" } + - { source_path: "rule.uuid", ocsf_field: "finding_info.analytic.uid" } + - { source_path: "rule.description", ocsf_field: "finding_info.analytic.desc" } + - { source_path: "rule.type", ocsf_field: "finding_info.analytic.type" } + # ── Incident state ──────────────────────────────────────────────── + - { source_path: "status.name", ocsf_field: "status" } + - { source_path: "alert_type.value", ocsf_field: "activity_name" } + - { source_path: "similar", ocsf_field: "count" } + - { source_path: "urgency.value", ocsf_field: "risk_score" } + # ── MITRE ATT&CK (first TTP / kill-chain step) ──────────────────── + - { source_path: "ttps[0].name", ocsf_field: "attacks.technique.name" } + - { source_path: "ttps[0].id", ocsf_field: "attacks.technique.uid" } + - { source_path: "kill_chain_short_id", ocsf_field: "attacks.tactic.uid" } + # ── Target host ─────────────────────────────────────────────────── + - { source_path: "target", ocsf_field: "src_endpoint.hostname" } + # ── Owning entity / tenant ──────────────────────────────────────── + - { source_path: "entity.name", ocsf_field: "cloud.org.name" } + - { source_path: "entity.uuid", ocsf_field: "cloud.org.uid" } + - { source_path: "short_id", ocsf_field: "metadata.uid" } + - { source_path: "community_uuid", ocsf_field: "metadata.tenant_uid" } diff --git a/integrations/sekoia-xdr/scripts/add_attributes_asset.py b/integrations/sekoia-xdr/scripts/add_attributes_asset.py new file mode 100644 index 0000000..dfdf3fc --- /dev/null +++ b/integrations/sekoia-xdr/scripts/add_attributes_asset.py @@ -0,0 +1,42 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path, params=None): + base, headers = _cfg() + url = base + "/" + path.lstrip("/") + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean, doseq=True) + req = urllib.request.Request(url, 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(): + inp = _inputs() + asset_uuid = urllib.parse.quote(inp.get("asset_uuid", ""), safe="") + params = {"name": inp.get("name"), "value": inp.get("value")} + print(json.dumps(request("POST", "/v1/asset-management/assets/" + asset_uuid + "/attr", params=params))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/add_keys_asset.py b/integrations/sekoia-xdr/scripts/add_keys_asset.py new file mode 100644 index 0000000..184544e --- /dev/null +++ b/integrations/sekoia-xdr/scripts/add_keys_asset.py @@ -0,0 +1,42 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path, params=None): + base, headers = _cfg() + url = base + "/" + path.lstrip("/") + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean, doseq=True) + req = urllib.request.Request(url, 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(): + inp = _inputs() + asset_uuid = urllib.parse.quote(inp.get("asset_uuid", ""), safe="") + params = {"name": inp.get("name"), "value": inp.get("value")} + print(json.dumps(request("POST", "/v1/asset-management/assets/" + asset_uuid + "/keys", params=params))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/events_execute_query.py b/integrations/sekoia-xdr/scripts/events_execute_query.py new file mode 100644 index 0000000..efa5630 --- /dev/null +++ b/integrations/sekoia-xdr/scripts/events_execute_query.py @@ -0,0 +1,39 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path, body=None): + base, headers = _cfg() + data = json.dumps(body).encode("utf-8") if body is not None else None + req = urllib.request.Request(base + "/" + path.lstrip("/"), 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(): + inp = _inputs() + body = {"earliest_time": inp.get("earliest_time"), "latest_time": inp.get("latest_time"), "term": inp.get("query")} + if inp.get("max_last_events"): + body["max_last_events"] = inp["max_last_events"] + print(json.dumps(request("POST", "/v1/sic/conf/events/search/jobs", body=body))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/events_results_query.py b/integrations/sekoia-xdr/scripts/events_results_query.py new file mode 100644 index 0000000..9ebf267 --- /dev/null +++ b/integrations/sekoia-xdr/scripts/events_results_query.py @@ -0,0 +1,35 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path): + base, headers = _cfg() + req = urllib.request.Request(base + "/" + path.lstrip("/"), 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(): + uuid = _inputs().get("uuid", "") + print(json.dumps(request("GET", "/v1/sic/conf/events/search/jobs/" + urllib.parse.quote(uuid, safe="") + "/events"))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/events_status_query.py b/integrations/sekoia-xdr/scripts/events_status_query.py new file mode 100644 index 0000000..946442c --- /dev/null +++ b/integrations/sekoia-xdr/scripts/events_status_query.py @@ -0,0 +1,35 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path): + base, headers = _cfg() + req = urllib.request.Request(base + "/" + path.lstrip("/"), 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(): + uuid = _inputs().get("uuid", "") + print(json.dumps(request("GET", "/v1/sic/conf/events/search/jobs/" + urllib.parse.quote(uuid, safe="")))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/get_alert.py b/integrations/sekoia-xdr/scripts/get_alert.py new file mode 100644 index 0000000..8715308 --- /dev/null +++ b/integrations/sekoia-xdr/scripts/get_alert.py @@ -0,0 +1,35 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path): + base, headers = _cfg() + req = urllib.request.Request(base + "/" + path.lstrip("/"), 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(): + alert_id = _inputs().get("id", "") + print(json.dumps(request("GET", "/v1/sic/alerts/" + urllib.parse.quote(alert_id, safe="")))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/get_asset.py b/integrations/sekoia-xdr/scripts/get_asset.py new file mode 100644 index 0000000..b49acbc --- /dev/null +++ b/integrations/sekoia-xdr/scripts/get_asset.py @@ -0,0 +1,35 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path): + base, headers = _cfg() + req = urllib.request.Request(base + "/" + path.lstrip("/"), 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(): + asset_uuid = urllib.parse.quote(_inputs().get("asset_uuid", ""), safe="") + print(json.dumps(request("GET", "/v1/asset-management/assets/" + asset_uuid))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/get_cases_alert.py b/integrations/sekoia-xdr/scripts/get_cases_alert.py new file mode 100644 index 0000000..e5b46b1 --- /dev/null +++ b/integrations/sekoia-xdr/scripts/get_cases_alert.py @@ -0,0 +1,43 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path, params=None): + base, headers = _cfg() + url = base + "/" + path.lstrip("/") + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean, doseq=True) + req = urllib.request.Request(url, 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(): + inp = _inputs() + params = {"match[alert_uuid]": inp.get("alert_id")} + if inp.get("case_id"): + params["match[short_id]"] = inp["case_id"] + print(json.dumps(request("GET", "/v1/sic/cases", params=params))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/get_comments.py b/integrations/sekoia-xdr/scripts/get_comments.py new file mode 100644 index 0000000..558e3cf --- /dev/null +++ b/integrations/sekoia-xdr/scripts/get_comments.py @@ -0,0 +1,35 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path): + base, headers = _cfg() + req = urllib.request.Request(base + "/" + path.lstrip("/"), 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(): + alert_id = urllib.parse.quote(_inputs().get("id", ""), safe="") + print(json.dumps(request("GET", "/v1/sic/alerts/" + alert_id + "/comments"))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/get_kill_chain.py b/integrations/sekoia-xdr/scripts/get_kill_chain.py new file mode 100644 index 0000000..2f1e7c0 --- /dev/null +++ b/integrations/sekoia-xdr/scripts/get_kill_chain.py @@ -0,0 +1,35 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path): + base, headers = _cfg() + req = urllib.request.Request(base + "/" + path.lstrip("/"), 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(): + kc = urllib.parse.quote(_inputs().get("kill_chain_uuid", ""), safe="") + print(json.dumps(request("GET", "/v1/sic/kill-chains/" + kc))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/get_user.py b/integrations/sekoia-xdr/scripts/get_user.py new file mode 100644 index 0000000..c710c7b --- /dev/null +++ b/integrations/sekoia-xdr/scripts/get_user.py @@ -0,0 +1,35 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path): + base, headers = _cfg() + req = urllib.request.Request(base + "/" + path.lstrip("/"), 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(): + user_uuid = urllib.parse.quote(_inputs().get("user_uuid", ""), safe="") + print(json.dumps(request("GET", "/v1/users/" + user_uuid))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/get_workflow_alert.py b/integrations/sekoia-xdr/scripts/get_workflow_alert.py new file mode 100644 index 0000000..ae94bf4 --- /dev/null +++ b/integrations/sekoia-xdr/scripts/get_workflow_alert.py @@ -0,0 +1,35 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path): + base, headers = _cfg() + req = urllib.request.Request(base + "/" + path.lstrip("/"), 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(): + alert_id = urllib.parse.quote(_inputs().get("id", ""), safe="") + print(json.dumps(request("GET", "/v1/sic/alerts/" + alert_id + "/workflow"))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/http_request.py b/integrations/sekoia-xdr/scripts/http_request.py new file mode 100644 index 0000000..6738100 --- /dev/null +++ b/integrations/sekoia-xdr/scripts/http_request.py @@ -0,0 +1,56 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _json_arg(value): + # Accept a JSON object string; ignore anything that does not parse to an object. + if not value: + return None + if isinstance(value, dict): + return value + try: + parsed = json.loads(value) + return parsed if isinstance(parsed, dict) else None + except Exception: + return None + + +def run(): + base, headers = _cfg() + inp = _inputs() + method = str(inp.get("method") or "GET").upper() + suffix = inp.get("url_suffix", "") + params = _json_arg(inp.get("parameters")) + body = _json_arg(inp.get("data")) + + url = base + "/" + str(suffix).lstrip("/") + if params: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(params, doseq=True) + data = json.dumps(body).encode("utf-8") if body is not None and method in ("POST", "PUT", "PATCH") else None + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=90) as r: + raw = r.read() + try: + print(json.dumps(json.loads(raw) if raw else {})) + except Exception: + print(json.dumps({"raw": raw.decode("utf-8", "replace")})) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/list_alerts.py b/integrations/sekoia-xdr/scripts/list_alerts.py new file mode 100644 index 0000000..c01382c --- /dev/null +++ b/integrations/sekoia-xdr/scripts/list_alerts.py @@ -0,0 +1,63 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path, params=None): + base, headers = _cfg() + url = base + "/" + path.lstrip("/") + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean, doseq=True) + req = urllib.request.Request(url, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=90) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _range(value): + # A bare date is the lower bound (",now"); a full range is passed through. + v = str(value) + return v if "," in v else (v + ",now") + + +def run(): + inp = _inputs() + params = { + "offset": inp.get("offset") or 0, + "direction": inp.get("direction") or "asc", + "sort": inp.get("sort_by") or "created_at", + } + if inp.get("limit"): + params["limit"] = inp["limit"] + if inp.get("status"): + params["match[status_name]"] = inp["status"] + if inp.get("created_at"): + params["date[created_at]"] = _range(inp["created_at"]) + if inp.get("updated_at"): + params["date[updated_at]"] = _range(inp["updated_at"]) + if inp.get("urgency"): + params["range[urgency]"] = inp["urgency"] + if inp.get("alerts_type"): + params["match[type_value]"] = inp["alerts_type"] + print(json.dumps(request("GET", "/v1/sic/alerts", params=params))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/list_assets.py b/integrations/sekoia-xdr/scripts/list_assets.py new file mode 100644 index 0000000..6aebfe2 --- /dev/null +++ b/integrations/sekoia-xdr/scripts/list_assets.py @@ -0,0 +1,43 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path, params=None): + base, headers = _cfg() + url = base + "/" + path.lstrip("/") + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean, doseq=True) + req = urllib.request.Request(url, 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(): + inp = _inputs() + params = {"limit": inp.get("limit") or 10} + if inp.get("assets_type"): + params["match[type_name]"] = inp["assets_type"] + print(json.dumps(request("GET", "/v1/asset-management/assets", params=params))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/post_comment_alert.py b/integrations/sekoia-xdr/scripts/post_comment_alert.py new file mode 100644 index 0000000..0d959d4 --- /dev/null +++ b/integrations/sekoia-xdr/scripts/post_comment_alert.py @@ -0,0 +1,40 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path, body=None): + base, headers = _cfg() + data = json.dumps(body).encode("utf-8") if body is not None else None + req = urllib.request.Request(base + "/" + path.lstrip("/"), 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(): + inp = _inputs() + alert_id = urllib.parse.quote(inp.get("id", ""), safe="") + body = {"content": inp.get("comment")} + if inp.get("author"): + body["author"] = inp["author"] + print(json.dumps(request("POST", "/v1/sic/alerts/" + alert_id + "/comments", body=body))) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/remove_attribute_asset.py b/integrations/sekoia-xdr/scripts/remove_attribute_asset.py new file mode 100644 index 0000000..6ace4f7 --- /dev/null +++ b/integrations/sekoia-xdr/scripts/remove_attribute_asset.py @@ -0,0 +1,37 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path): + base, headers = _cfg() + req = urllib.request.Request(base + "/" + path.lstrip("/"), headers=headers, method=method) + with urllib.request.urlopen(req, timeout=90) as r: + return r.read().decode("utf-8", "replace") + + +def run(): + inp = _inputs() + asset_uuid = urllib.parse.quote(inp.get("asset_uuid", ""), safe="") + attribute_uuid = urllib.parse.quote(inp.get("attribute_uuid", ""), safe="") + request("DELETE", "/v1/asset-management/assets/" + asset_uuid + "/attr/" + attribute_uuid) + print(json.dumps({"removed": True, "asset_uuid": inp.get("asset_uuid"), "attribute_uuid": inp.get("attribute_uuid")})) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/remove_key_asset.py b/integrations/sekoia-xdr/scripts/remove_key_asset.py new file mode 100644 index 0000000..6459542 --- /dev/null +++ b/integrations/sekoia-xdr/scripts/remove_key_asset.py @@ -0,0 +1,37 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path): + base, headers = _cfg() + req = urllib.request.Request(base + "/" + path.lstrip("/"), headers=headers, method=method) + with urllib.request.urlopen(req, timeout=90) as r: + return r.read().decode("utf-8", "replace") + + +def run(): + inp = _inputs() + asset_uuid = urllib.parse.quote(inp.get("asset_uuid", ""), safe="") + key_uuid = urllib.parse.quote(inp.get("key_uuid", ""), safe="") + request("DELETE", "/v1/asset-management/assets/" + asset_uuid + "/keys/" + key_uuid) + print(json.dumps({"removed": True, "asset_uuid": inp.get("asset_uuid"), "key_uuid": inp.get("key_uuid")})) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/search_events.py b/integrations/sekoia-xdr/scripts/search_events.py new file mode 100644 index 0000000..980f0af --- /dev/null +++ b/integrations/sekoia-xdr/scripts/search_events.py @@ -0,0 +1,50 @@ +import json, os, sys, time, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path, body=None): + base, headers = _cfg() + data = json.dumps(body).encode("utf-8") if body is not None else None + req = urllib.request.Request(base + "/" + path.lstrip("/"), 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(): + inp = _inputs() + body = {"earliest_time": inp.get("earliest_time"), "latest_time": inp.get("latest_time"), "term": inp.get("query")} + if inp.get("max_last_events"): + body["max_last_events"] = inp["max_last_events"] + job = request("POST", "/v1/sic/conf/events/search/jobs", body=body) + job_uuid = job.get("uuid") + # Poll the job until it finishes (status == 2), bounded to stay within the run timeout. + for _ in range(24): + status = request("GET", "/v1/sic/conf/events/search/jobs/" + urllib.parse.quote(job_uuid, safe="")) + if status.get("status") == 2: + events = request("GET", "/v1/sic/conf/events/search/jobs/" + urllib.parse.quote(job_uuid, safe="") + "/events") + print(json.dumps(events)) + return + time.sleep(2) + # Still running: hand back the job UUID so the caller can retrieve results later. + print(json.dumps({"pending": True, "job_uuid": job_uuid, "message": "Event search still running; retrieve with events_results_query."})) + + +try: + run() +except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) +except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/test_connection.py b/integrations/sekoia-xdr/scripts/test_connection.py new file mode 100644 index 0000000..8e2e740 --- /dev/null +++ b/integrations/sekoia-xdr/scripts/test_connection.py @@ -0,0 +1,33 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Accept": "application/json"} + return base, headers + + +def request(method, path): + base, headers = _cfg() + req = urllib.request.Request(base + "/" + path.lstrip("/"), headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def run(): + request("GET", "/v1/auth/validate") + print(json.dumps({"ok": True})) + + +try: + run() +except urllib.error.HTTPError as e: + detail = e.read().decode("utf-8", "replace") + msg = "API key is not valid." if e.code in (401, 403) else "HTTP " + str(e.code) + print(json.dumps({"ok": False, "error": msg, "detail": detail})) + sys.exit(1) +except Exception as e: + print(json.dumps({"ok": False, "error": str(e)})) + sys.exit(1) diff --git a/integrations/sekoia-xdr/scripts/update_status_alert.py b/integrations/sekoia-xdr/scripts/update_status_alert.py new file mode 100644 index 0000000..8f8766f --- /dev/null +++ b/integrations/sekoia-xdr/scripts/update_status_alert.py @@ -0,0 +1,52 @@ +import json, os, sys, urllib.request, urllib.parse, urllib.error + +# Map a target status to the Sekoia workflow transition name. +TRANSITIONS = {"Ongoing": "Validate", "Acknowledged": "Acknowledge", "Rejected": "Reject", "Closed": "Close"} + + +def _cfg(): + s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + base = str(s.get("url") or "https://api.sekoia.io").rstrip("/") + headers = {"Authorization": "Bearer " + s.get("api_key", ""), "Content-Type": "application/json", "Accept": "application/json"} + return base, headers + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(method, path, body=None): + base, headers = _cfg() + data = json.dumps(body).encode("utf-8") if body is not None else None + req = urllib.request.Request(base + "/" + path.lstrip("/"), 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(): + inp = _inputs() + alert_id = urllib.parse.quote(inp.get("id", ""), safe="") + status = inp.get("status", "") + transition = TRANSITIONS.get(status) + workflow = request("GET", "/v1/sic/alerts/" + alert_id + "/workflow") + action = next((a for a in workflow.get("actions", []) if a.get("name") == transition), None) + if not action: + print(json.dumps({"error": "No workflow transition available for status '" + status + "'.", + "available": [a.get("name") for a in workflow.get("actions", [])]})) + sys.exit(1) + body = {"action_uuid": action["id"]} + if inp.get("comment"): + body["comment"] = inp["comment"] + result = request("PATCH", "/v1/sic/alerts/" + alert_id + "/workflow", body=body) + print(json.dumps({"updated": True, "status": status, "result": result})) + + +try: + run() +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)