From 2b3b9fc10c12e44e53d378c1a50ce9ef7e453067 Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sat, 11 Jul 2026 22:33:10 +0200 Subject: [PATCH] feat(alienvault-otx): new AlienVault OTX enrichment integration 9 commands: IP/domain/URL/file reputation, pulse details + search, passive DNS and related URLs. API-key auth, stdlib-only. Co-Authored-By: Claude Fable 5 --- integrations/alienvault-otx/manifest.yaml | 101 ++++++++++++++++++ .../scripts/domain_reputation.py | 41 +++++++ .../alienvault-otx/scripts/file_reputation.py | 41 +++++++ .../alienvault-otx/scripts/get_pulse.py | 41 +++++++ .../alienvault-otx/scripts/ip_reputation.py | 43 ++++++++ .../alienvault-otx/scripts/passive_dns.py | 44 ++++++++ .../alienvault-otx/scripts/related_urls.py | 44 ++++++++ .../alienvault-otx/scripts/search_pulses.py | 42 ++++++++ .../alienvault-otx/scripts/test_connection.py | 39 +++++++ .../alienvault-otx/scripts/url_reputation.py | 41 +++++++ 10 files changed, 477 insertions(+) create mode 100644 integrations/alienvault-otx/manifest.yaml create mode 100644 integrations/alienvault-otx/scripts/domain_reputation.py create mode 100644 integrations/alienvault-otx/scripts/file_reputation.py create mode 100644 integrations/alienvault-otx/scripts/get_pulse.py create mode 100644 integrations/alienvault-otx/scripts/ip_reputation.py create mode 100644 integrations/alienvault-otx/scripts/passive_dns.py create mode 100644 integrations/alienvault-otx/scripts/related_urls.py create mode 100644 integrations/alienvault-otx/scripts/search_pulses.py create mode 100644 integrations/alienvault-otx/scripts/test_connection.py create mode 100644 integrations/alienvault-otx/scripts/url_reputation.py diff --git a/integrations/alienvault-otx/manifest.yaml b/integrations/alienvault-otx/manifest.yaml new file mode 100644 index 0000000..529691c --- /dev/null +++ b/integrations/alienvault-otx/manifest.yaml @@ -0,0 +1,101 @@ +id: alienvault_otx +name: AlienVault OTX +version: 1.0.0 +description: "AlienVault OTX (Open Threat Exchange, API v1) — reputation and threat context for IPs, domains, URLs and file hashes, pulse details and search, and passive DNS / related-URL pivots. API-key authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: IP/domain/URL/file reputation, pulse details and search, passive DNS and related URLs." +category: enrichment + +config_schema: + properties: + api_key: + type: string + description: "AlienVault OTX API key (from your OTX account settings)" + x-soar-sensitive: true + required: + - api_key + +commands: + - id: ip_reputation + name: alienvault-otx-ip + description: "Threat context for an IP address (IPv4 or IPv6)." + risk: read + inputs_schema: + properties: + ip: { type: string, description: "IP address" } + required: [ip] + outputs_schema: { properties: {} } + - id: domain_reputation + name: alienvault-otx-domain + description: "Threat context for a domain." + risk: read + inputs_schema: + properties: + domain: { type: string, description: "Domain name" } + required: [domain] + outputs_schema: { properties: {} } + - id: url_reputation + name: alienvault-otx-url + description: "Threat context for a URL." + risk: read + inputs_schema: + properties: + url: { type: string, description: "URL" } + required: [url] + outputs_schema: { properties: {} } + - id: file_reputation + name: alienvault-otx-file + description: "Threat context for a file hash (MD5, SHA1 or SHA256)." + risk: read + inputs_schema: + properties: + file: { type: string, description: "File hash" } + required: [file] + outputs_schema: { properties: {} } + - id: get_pulse + name: alienvault-otx-get-pulse + description: "Get the details of a pulse (threat report) by ID." + risk: read + inputs_schema: + properties: + pulse_id: { type: string, description: "Pulse ID" } + required: [pulse_id] + outputs_schema: { properties: {} } + - id: search_pulses + name: alienvault-otx-search-pulses + description: "Search pulses by keyword." + risk: read + inputs_schema: + properties: + query: { type: string, description: "Search string" } + limit: { type: number, description: "Maximum pulses (default 20)" } + required: [query] + outputs_schema: { properties: {} } + - id: passive_dns + name: alienvault-otx-passive-dns + description: "Passive DNS records for an IP or domain indicator." + risk: read + inputs_schema: + properties: + indicator: { type: string, description: "IP or domain" } + indicator_type: { type: string, description: "IPv4, IPv6 or domain (default auto-detected)" } + required: [indicator] + outputs_schema: { properties: {} } + - id: related_urls + name: alienvault-otx-related-urls + description: "URLs associated with an IP or domain indicator." + risk: read + inputs_schema: + properties: + indicator: { type: string, description: "IP or domain" } + indicator_type: { type: string, description: "IPv4, IPv6 or domain (default auto-detected)" } + required: [indicator] + outputs_schema: { properties: {} } + + - id: test_connection + name: alienvault-otx-test-connection + description: "Verify the API key (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/alienvault-otx/scripts/domain_reputation.py b/integrations/alienvault-otx/scripts/domain_reputation.py new file mode 100644 index 0000000..06bfa0c --- /dev/null +++ b/integrations/alienvault-otx/scripts/domain_reputation.py @@ -0,0 +1,41 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://otx.alienvault.com/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")} + req = urllib.request.Request(url, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + domain = inputs.get("domain") + if not domain: + raise Exception("domain is required") + res = request("GET", "/indicators/domain/%s/general" % q(domain)) + print(json.dumps(res)) + + +try: + main() +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/alienvault-otx/scripts/file_reputation.py b/integrations/alienvault-otx/scripts/file_reputation.py new file mode 100644 index 0000000..ae8f53c --- /dev/null +++ b/integrations/alienvault-otx/scripts/file_reputation.py @@ -0,0 +1,41 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://otx.alienvault.com/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")} + req = urllib.request.Request(url, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + file_hash = inputs.get("file") + if not file_hash: + raise Exception("file is required") + res = request("GET", "/indicators/file/%s/general" % q(file_hash)) + print(json.dumps(res)) + + +try: + main() +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/alienvault-otx/scripts/get_pulse.py b/integrations/alienvault-otx/scripts/get_pulse.py new file mode 100644 index 0000000..c99ab2e --- /dev/null +++ b/integrations/alienvault-otx/scripts/get_pulse.py @@ -0,0 +1,41 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://otx.alienvault.com/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")} + req = urllib.request.Request(url, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + pulse_id = inputs.get("pulse_id") + if not pulse_id: + raise Exception("pulse_id is required") + res = request("GET", "/pulses/%s" % q(pulse_id)) + print(json.dumps(res)) + + +try: + main() +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/alienvault-otx/scripts/ip_reputation.py b/integrations/alienvault-otx/scripts/ip_reputation.py new file mode 100644 index 0000000..97df7e6 --- /dev/null +++ b/integrations/alienvault-otx/scripts/ip_reputation.py @@ -0,0 +1,43 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error +import re + +API = "https://otx.alienvault.com/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")} + req = urllib.request.Request(url, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + ip = inputs.get("ip") + if not ip: + raise Exception("ip is required") + section = "IPv6" if ":" in ip else "IPv4" + res = request("GET", "/indicators/%s/%s/general" % (section, q(ip))) + print(json.dumps(res)) + + +try: + main() +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/alienvault-otx/scripts/passive_dns.py b/integrations/alienvault-otx/scripts/passive_dns.py new file mode 100644 index 0000000..b3be38c --- /dev/null +++ b/integrations/alienvault-otx/scripts/passive_dns.py @@ -0,0 +1,44 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error +import re + +API = "https://otx.alienvault.com/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")} + req = urllib.request.Request(url, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + indicator = inputs.get("indicator") + if not indicator: + raise Exception("indicator is required") + indicator_type = inputs.get("indicator_type") + itype = indicator_type or ("IPv6" if ":" in indicator else ("IPv4" if re.match(r"^\d+\.\d+\.\d+\.\d+$", indicator) else "domain")) + res = request("GET", "/indicators/%s/%s/passive_dns" % (itype, q(indicator))) + print(json.dumps(res)) + + +try: + main() +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/alienvault-otx/scripts/related_urls.py b/integrations/alienvault-otx/scripts/related_urls.py new file mode 100644 index 0000000..ad5f4d1 --- /dev/null +++ b/integrations/alienvault-otx/scripts/related_urls.py @@ -0,0 +1,44 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error +import re + +API = "https://otx.alienvault.com/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")} + req = urllib.request.Request(url, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + indicator = inputs.get("indicator") + if not indicator: + raise Exception("indicator is required") + indicator_type = inputs.get("indicator_type") + itype = indicator_type or ("IPv6" if ":" in indicator else ("IPv4" if re.match(r"^\d+\.\d+\.\d+\.\d+$", indicator) else "domain")) + res = request("GET", "/indicators/%s/%s/url_list" % (itype, q(indicator)), {"limit": 100}) + print(json.dumps(res)) + + +try: + main() +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/alienvault-otx/scripts/search_pulses.py b/integrations/alienvault-otx/scripts/search_pulses.py new file mode 100644 index 0000000..3a375ee --- /dev/null +++ b/integrations/alienvault-otx/scripts/search_pulses.py @@ -0,0 +1,42 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://otx.alienvault.com/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")} + req = urllib.request.Request(url, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + query = inputs.get("query") + if not query: + raise Exception("query is required") + limit = inputs.get("limit") + res = request("GET", "/search/pulses", {"q": query, "limit": limit or 20}) + print(json.dumps(res)) + + +try: + main() +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/alienvault-otx/scripts/test_connection.py b/integrations/alienvault-otx/scripts/test_connection.py new file mode 100644 index 0000000..e16020f --- /dev/null +++ b/integrations/alienvault-otx/scripts/test_connection.py @@ -0,0 +1,39 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://otx.alienvault.com/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")} + req = urllib.request.Request(url, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + res = request("GET", "/user/me") + if "username" not in res: + raise Exception("unexpected response") + print(json.dumps({"ok": True, "user": res.get("username")})) + + +try: + main() +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/alienvault-otx/scripts/url_reputation.py b/integrations/alienvault-otx/scripts/url_reputation.py new file mode 100644 index 0000000..b60cbd1 --- /dev/null +++ b/integrations/alienvault-otx/scripts/url_reputation.py @@ -0,0 +1,41 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://otx.alienvault.com/api/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(method, path, params=None): + url = API + path + p = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")} + if p: + url += ("&" if "?" in url else "?") + urllib.parse.urlencode(p) + headers = {"Accept": "application/json", "X-OTX-API-KEY": str(_cfg().get("api_key") or "")} + req = urllib.request.Request(url, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + url = inputs.get("url") + if not url: + raise Exception("url is required") + res = request("GET", "/indicators/url/%s/general" % q(url)) + print(json.dumps(res)) + + +try: + main() +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)