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 <noreply@anthropic.com>

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: {} }

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)

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)

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)

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)

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)

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)

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)

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)

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)