feat(securitytrails): new SecurityTrails enrichment integration
8 commands: domain details, subdomains, WHOIS, DNS history, associated domains, IP neighbors, domain search. API-key auth, stdlib-only. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -0,0 +1,93 @@
|
||||
id: securitytrails
|
||||
name: SecurityTrails
|
||||
version: 1.0.0
|
||||
description: "SecurityTrails (API v1) — DNS and domain intelligence: current domain details, subdomains, WHOIS, historical DNS, associated domains, IP neighbors and a domain filter search. API-key authentication; stdlib-only, no extra Python dependencies."
|
||||
changelog: "1.0.0 — Initial release: domain details, subdomains, WHOIS, DNS history, associated domains, IP neighbors and domain search."
|
||||
category: enrichment
|
||||
|
||||
config_schema:
|
||||
properties:
|
||||
api_key:
|
||||
type: string
|
||||
description: "SecurityTrails API key"
|
||||
x-soar-sensitive: true
|
||||
required:
|
||||
- api_key
|
||||
|
||||
commands:
|
||||
- id: domain_details
|
||||
name: securitytrails-domain-details
|
||||
description: "Current DNS and metadata for a domain (A/MX/NS/SOA/TXT records, host provider, Alexa rank)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
domain: { type: string, description: "Domain name" }
|
||||
required: [domain]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: subdomains
|
||||
name: securitytrails-subdomains
|
||||
description: "List the known subdomains of a domain."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
domain: { type: string, description: "Domain name" }
|
||||
required: [domain]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: domain_whois
|
||||
name: securitytrails-domain-whois
|
||||
description: "Current WHOIS record for a domain."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
domain: { type: string, description: "Domain name" }
|
||||
required: [domain]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: dns_history
|
||||
name: securitytrails-dns-history
|
||||
description: "Historical DNS records of a domain for a given record type."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
domain: { type: string, description: "Domain name" }
|
||||
record_type: { type: string, description: "Record type: a, aaaa, mx, ns, soa or txt (default a)" }
|
||||
required: [domain]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: associated_domains
|
||||
name: securitytrails-associated-domains
|
||||
description: "Domains associated with the given domain (shared registrant/infrastructure)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
domain: { type: string, description: "Domain name" }
|
||||
required: [domain]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: ip_neighbors
|
||||
name: securitytrails-ip-neighbors
|
||||
description: "Neighboring IPs in the same range as the given IP, with hostname counts."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
ip: { type: string, description: "IP address" }
|
||||
required: [ip]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: search_domain
|
||||
name: securitytrails-search-domain
|
||||
description: "Search domains with a filter (e.g. by keyword, mail provider, whois email)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
keyword: { type: string, description: "Substring the domain must contain" }
|
||||
mail_provider: { type: string, description: "Mail provider filter" }
|
||||
whois_email: { type: string, description: "WHOIS registrant email filter" }
|
||||
limit: { type: number, description: "Maximum records (default 100)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
- id: test_connection
|
||||
name: securitytrails-test-connection
|
||||
description: "Verify the API key (used by the Test button)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
@@ -0,0 +1,44 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.securitytrails.com/v1"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, params=None, body=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)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "APIKEY": str(_cfg().get("api_key") or "")}
|
||||
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 {}
|
||||
|
||||
|
||||
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", "/domain/" + q(domain) + "/associated")
|
||||
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)
|
||||
@@ -0,0 +1,45 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.securitytrails.com/v1"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, params=None, body=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)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "APIKEY": str(_cfg().get("api_key") or "")}
|
||||
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 {}
|
||||
|
||||
|
||||
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")
|
||||
record_type = inputs.get("record_type") or "a"
|
||||
res = request("GET", "/history/" + q(domain) + "/dns/" + q(record_type))
|
||||
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)
|
||||
@@ -0,0 +1,44 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.securitytrails.com/v1"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, params=None, body=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)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "APIKEY": str(_cfg().get("api_key") or "")}
|
||||
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 {}
|
||||
|
||||
|
||||
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", "/domain/" + 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)
|
||||
@@ -0,0 +1,44 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.securitytrails.com/v1"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, params=None, body=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)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "APIKEY": str(_cfg().get("api_key") or "")}
|
||||
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 {}
|
||||
|
||||
|
||||
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", "/domain/" + q(domain) + "/whois")
|
||||
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)
|
||||
@@ -0,0 +1,44 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.securitytrails.com/v1"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, params=None, body=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)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "APIKEY": str(_cfg().get("api_key") or "")}
|
||||
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 {}
|
||||
|
||||
|
||||
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")
|
||||
res = request("GET", "/ips/nearby/" + 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)
|
||||
@@ -0,0 +1,56 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.securitytrails.com/v1"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, params=None, body=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)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "APIKEY": str(_cfg().get("api_key") or "")}
|
||||
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 {}
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
keyword = inputs.get("keyword")
|
||||
mail_provider = inputs.get("mail_provider")
|
||||
whois_email = inputs.get("whois_email")
|
||||
limit = inputs.get("limit")
|
||||
|
||||
filt = {}
|
||||
if keyword:
|
||||
filt["keyword"] = keyword
|
||||
if mail_provider:
|
||||
filt["mail_provider"] = mail_provider
|
||||
if whois_email:
|
||||
filt["whois_email"] = whois_email
|
||||
if not filt:
|
||||
raise Exception("provide at least one filter (keyword, mail_provider or whois_email)")
|
||||
|
||||
res = request("POST", "/domains/list", params={"limit": limit or 100}, body={"filter": filt})
|
||||
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)
|
||||
@@ -0,0 +1,44 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.securitytrails.com/v1"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, params=None, body=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)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "APIKEY": str(_cfg().get("api_key") or "")}
|
||||
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 {}
|
||||
|
||||
|
||||
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", "/domain/" + q(domain) + "/subdomains")
|
||||
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)
|
||||
@@ -0,0 +1,42 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.securitytrails.com/v1"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, params=None, body=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)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Accept": "application/json", "APIKEY": str(_cfg().get("api_key") or "")}
|
||||
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 {}
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
res = request("GET", "/ping")
|
||||
if not (isinstance(res, dict) and res.get("success")):
|
||||
raise Exception("unexpected response")
|
||||
print(json.dumps({"ok": True}))
|
||||
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user