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

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)

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)

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)

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)

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)

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)

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)

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)

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)