From a5db72552996b7e3dca8d04925c756a6867f4d3e Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sun, 12 Jul 2026 00:15:11 +0200 Subject: [PATCH] feat(anomali-threatstream): new Anomali ThreatStream threat-intel integration ThreatStream API, 9 commands: ip/domain/file/url reputation, indicator search, passive DNS, threat models, import indicator. apikey auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../anomali-threatstream/manifest.yaml | 112 ++++++++++++++++++ .../scripts/domain_reputation.py | 54 +++++++++ .../scripts/file_reputation.py | 54 +++++++++ .../scripts/get_indicators.py | 53 +++++++++ .../scripts/get_threat_models.py | 53 +++++++++ .../scripts/import_indicator.py | 63 ++++++++++ .../scripts/ip_reputation.py | 54 +++++++++ .../scripts/passive_dns.py | 57 +++++++++ .../scripts/test_connection.py | 52 ++++++++ .../scripts/url_reputation.py | 54 +++++++++ 10 files changed, 606 insertions(+) create mode 100644 integrations/anomali-threatstream/manifest.yaml create mode 100644 integrations/anomali-threatstream/scripts/domain_reputation.py create mode 100644 integrations/anomali-threatstream/scripts/file_reputation.py create mode 100644 integrations/anomali-threatstream/scripts/get_indicators.py create mode 100644 integrations/anomali-threatstream/scripts/get_threat_models.py create mode 100644 integrations/anomali-threatstream/scripts/import_indicator.py create mode 100644 integrations/anomali-threatstream/scripts/ip_reputation.py create mode 100644 integrations/anomali-threatstream/scripts/passive_dns.py create mode 100644 integrations/anomali-threatstream/scripts/test_connection.py create mode 100644 integrations/anomali-threatstream/scripts/url_reputation.py diff --git a/integrations/anomali-threatstream/manifest.yaml b/integrations/anomali-threatstream/manifest.yaml new file mode 100644 index 0000000..3a97b40 --- /dev/null +++ b/integrations/anomali-threatstream/manifest.yaml @@ -0,0 +1,112 @@ +id: anomali_threatstream +name: Anomali ThreatStream +version: 1.0.0 +description: "Anomali ThreatStream (API v2/v1) — threat intelligence: reputation lookups for IPs, domains, file hashes and URLs, indicator search, passive DNS, threat-model listing, and indicator import (with or without approval). API-key authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: ip/domain/file/url reputation, indicator search, passive DNS, threat models, import indicator." +category: threat_intel + +# Per-instance configuration. Auth header 'Authorization: apikey :'. +config_schema: + properties: + url: + type: string + description: "ThreatStream API URL" + default: "https://api.threatstream.com" + username: + type: string + description: "ThreatStream username" + api_key: + type: string + description: "ThreatStream API key" + x-soar-sensitive: true + required: + - username + - api_key + +commands: + - id: ip_reputation + name: anomali-ip-reputation + description: "Look up threat intelligence for an IP address." + risk: read + inputs_schema: + properties: + ip: { type: string, description: "IP address" } + required: [ip] + outputs_schema: { properties: {} } + - id: domain_reputation + name: anomali-domain-reputation + description: "Look up threat intelligence for a domain." + risk: read + inputs_schema: + properties: + domain: { type: string, description: "Domain name" } + required: [domain] + outputs_schema: { properties: {} } + - id: file_reputation + name: anomali-file-reputation + description: "Look up threat intelligence for a file hash." + risk: read + inputs_schema: + properties: + file_hash: { type: string, description: "MD5/SHA1/SHA256 hash" } + required: [file_hash] + outputs_schema: { properties: {} } + - id: url_reputation + name: anomali-url-reputation + description: "Look up threat intelligence for a URL." + risk: read + inputs_schema: + properties: + url: { type: string, description: "URL" } + required: [url] + outputs_schema: { properties: {} } + - id: get_indicators + name: anomali-get-indicators + description: "Search indicators with a free-text query." + risk: read + inputs_schema: + properties: + query: { type: string, description: "ThreatStream search query (q=)" } + limit: { type: number, description: "Max indicators (default 20)" } + required: [] + outputs_schema: { properties: {} } + - id: passive_dns + name: anomali-passive-dns + description: "Get passive DNS records for an IP or domain." + risk: read + inputs_schema: + properties: + value: { type: string, description: "IP or domain" } + type: { type: string, description: "ip or domain (default ip)" } + required: [value] + outputs_schema: { properties: {} } + - id: get_threat_models + name: anomali-get-threat-models + description: "List threat models (actors, campaigns, incidents, ...)." + risk: read + inputs_schema: + properties: + query: { type: string, description: "Optional name search" } + limit: { type: number, description: "Max models (default 20)" } + required: [] + outputs_schema: { properties: {} } + - id: import_indicator + name: anomali-import-indicator + description: "Import an observable as an indicator (optionally requiring approval)." + inputs_schema: + properties: + value: { type: string, description: "Observable value (IP, domain, hash, URL)" } + itype: { type: string, description: "Indicator type (e.g. mal_ip, mal_domain, apt_md5)" } + confidence: { type: number, description: "Confidence 0-100 (default 50)" } + approve: { type: boolean, description: "Import without approval (default false = requires approval)" } + required: [value, itype] + outputs_schema: { properties: {} } + + - id: test_connection + name: anomali-test-connection + description: "Verify connectivity and the API key (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/anomali-threatstream/scripts/domain_reputation.py b/integrations/anomali-threatstream/scripts/domain_reputation.py new file mode 100644 index 0000000..495079f --- /dev/null +++ b/integrations/anomali-threatstream/scripts/domain_reputation.py @@ -0,0 +1,54 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("url") or "https://api.threatstream.com")).rstrip("/") + + +def _auth(cfg): + return "apikey " + str(cfg.get("username", "")) + ":" + str(cfg.get("api_key", "")) + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, 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(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + domain = inputs.get("domain") + if not domain: + raise Exception("domain is required") + return request("GET", "/api/v2/intelligence/", cfg, params={"value": domain, "type": "domain", "limit": 50}) + + +_run(main) diff --git a/integrations/anomali-threatstream/scripts/file_reputation.py b/integrations/anomali-threatstream/scripts/file_reputation.py new file mode 100644 index 0000000..0ed76ea --- /dev/null +++ b/integrations/anomali-threatstream/scripts/file_reputation.py @@ -0,0 +1,54 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("url") or "https://api.threatstream.com")).rstrip("/") + + +def _auth(cfg): + return "apikey " + str(cfg.get("username", "")) + ":" + str(cfg.get("api_key", "")) + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, 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(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + file_hash = inputs.get("file_hash") + if not file_hash: + raise Exception("file_hash is required") + return request("GET", "/api/v2/intelligence/", cfg, params={"value": file_hash, "type": "md5", "limit": 50}) + + +_run(main) diff --git a/integrations/anomali-threatstream/scripts/get_indicators.py b/integrations/anomali-threatstream/scripts/get_indicators.py new file mode 100644 index 0000000..a75be90 --- /dev/null +++ b/integrations/anomali-threatstream/scripts/get_indicators.py @@ -0,0 +1,53 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("url") or "https://api.threatstream.com")).rstrip("/") + + +def _auth(cfg): + return "apikey " + str(cfg.get("username", "")) + ":" + str(cfg.get("api_key", "")) + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, 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(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + query = inputs.get("query") + limit = inputs.get("limit") + return request("GET", "/api/v2/intelligence/", cfg, params={"q": query, "limit": int(limit or 20)}) + + +_run(main) diff --git a/integrations/anomali-threatstream/scripts/get_threat_models.py b/integrations/anomali-threatstream/scripts/get_threat_models.py new file mode 100644 index 0000000..041d037 --- /dev/null +++ b/integrations/anomali-threatstream/scripts/get_threat_models.py @@ -0,0 +1,53 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("url") or "https://api.threatstream.com")).rstrip("/") + + +def _auth(cfg): + return "apikey " + str(cfg.get("username", "")) + ":" + str(cfg.get("api_key", "")) + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, 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(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + query = inputs.get("query") + limit = inputs.get("limit") + return request("GET", "/api/v1/threat_model_search/", cfg, params={"name": query, "limit": int(limit or 20)}) + + +_run(main) diff --git a/integrations/anomali-threatstream/scripts/import_indicator.py b/integrations/anomali-threatstream/scripts/import_indicator.py new file mode 100644 index 0000000..1d2e402 --- /dev/null +++ b/integrations/anomali-threatstream/scripts/import_indicator.py @@ -0,0 +1,63 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("url") or "https://api.threatstream.com")).rstrip("/") + + +def _auth(cfg): + return "apikey " + str(cfg.get("username", "")) + ":" + str(cfg.get("api_key", "")) + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, 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(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + value = inputs.get("value") + if not value: + raise Exception("value is required") + itype = inputs.get("itype") + if not itype: + raise Exception("itype is required") + confidence = inputs.get("confidence") + approve = inputs.get("approve") + body = {"objects": [{"value": value, "itype": itype, "confidence": int(confidence or 50)}]} + params = {} + if approve: + params["approve"] = "true" + return request("POST", "/api/v2/intelligence/", cfg, body=body, params=params or None) + + +_run(main) diff --git a/integrations/anomali-threatstream/scripts/ip_reputation.py b/integrations/anomali-threatstream/scripts/ip_reputation.py new file mode 100644 index 0000000..a9c978b --- /dev/null +++ b/integrations/anomali-threatstream/scripts/ip_reputation.py @@ -0,0 +1,54 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("url") or "https://api.threatstream.com")).rstrip("/") + + +def _auth(cfg): + return "apikey " + str(cfg.get("username", "")) + ":" + str(cfg.get("api_key", "")) + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, 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(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + ip = inputs.get("ip") + if not ip: + raise Exception("ip is required") + return request("GET", "/api/v2/intelligence/", cfg, params={"value": ip, "type": "ip", "limit": 50}) + + +_run(main) diff --git a/integrations/anomali-threatstream/scripts/passive_dns.py b/integrations/anomali-threatstream/scripts/passive_dns.py new file mode 100644 index 0000000..c29a1a3 --- /dev/null +++ b/integrations/anomali-threatstream/scripts/passive_dns.py @@ -0,0 +1,57 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("url") or "https://api.threatstream.com")).rstrip("/") + + +def _auth(cfg): + return "apikey " + str(cfg.get("username", "")) + ":" + str(cfg.get("api_key", "")) + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, 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(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + q = lambda v: urllib.parse.quote(str(v), safe="") + value = inputs.get("value") + if not value: + raise Exception("value is required") + itype = inputs.get("type") or "ip" + path = "/api/v1/pdns/" + q(itype) + "/" + q(value) + "/" + return request("GET", path, cfg) + + +_run(main) diff --git a/integrations/anomali-threatstream/scripts/test_connection.py b/integrations/anomali-threatstream/scripts/test_connection.py new file mode 100644 index 0000000..054fc9d --- /dev/null +++ b/integrations/anomali-threatstream/scripts/test_connection.py @@ -0,0 +1,52 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("url") or "https://api.threatstream.com")).rstrip("/") + + +def _auth(cfg): + return "apikey " + str(cfg.get("username", "")) + ":" + str(cfg.get("api_key", "")) + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, 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(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + request("GET", "/api/v2/intelligence/", cfg, params={"limit": 1}) + return {"ok": True} + + +_run(main) diff --git a/integrations/anomali-threatstream/scripts/url_reputation.py b/integrations/anomali-threatstream/scripts/url_reputation.py new file mode 100644 index 0000000..459da30 --- /dev/null +++ b/integrations/anomali-threatstream/scripts/url_reputation.py @@ -0,0 +1,54 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def _base(cfg): + return (str(cfg.get("url") or "https://api.threatstream.com")).rstrip("/") + + +def _auth(cfg): + return "apikey " + str(cfg.get("username", "")) + ":" + str(cfg.get("api_key", "")) + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, 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(fn): + try: + print(json.dumps(fn(_cfg(), _inputs()))) + except urllib.error.HTTPError as e: + print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")})) + sys.exit(1) + except Exception as e: + print(json.dumps({"error": str(e)})) + sys.exit(1) + + +def main(cfg, inputs): + url = inputs.get("url") + if not url: + raise Exception("url is required") + return request("GET", "/api/v2/intelligence/", cfg, params={"value": url, "type": "url", "limit": 50}) + + +_run(main)