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) <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -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 <username>:<api_key>'.
|
||||
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: {} }
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user