diff --git a/integrations/domaintools/manifest.yaml b/integrations/domaintools/manifest.yaml new file mode 100644 index 0000000..54c5179 --- /dev/null +++ b/integrations/domaintools/manifest.yaml @@ -0,0 +1,67 @@ +id: domaintools +name: DomainTools +version: 1.0.0 +description: "DomainTools (Iris/Enterprise API) — domain and DNS intelligence: WHOIS lookup and history, domain reputation and risk scoring, and reverse-IP host domains. API-key authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: whois, whois history, reputation, reverse IP." +category: enrichment + +# Per-instance configuration. api_username + api_key are sent as query parameters. +config_schema: + properties: + api_username: + type: string + description: "DomainTools API username" + api_key: + type: string + description: "DomainTools API key" + x-soar-sensitive: true + required: + - api_username + - api_key + +commands: + - id: whois + name: domaintools-whois + description: "Get current WHOIS for a domain." + risk: read + inputs_schema: + properties: + domain: { type: string, description: "Domain name" } + required: [domain] + outputs_schema: { properties: {} } + - id: whois_history + name: domaintools-whois-history + description: "Get historical WHOIS records for a domain." + risk: read + inputs_schema: + properties: + domain: { type: string, description: "Domain name" } + required: [domain] + outputs_schema: { properties: {} } + - id: reputation + name: domaintools-reputation + description: "Get a domain's reputation/risk score." + risk: read + inputs_schema: + properties: + domain: { type: string, description: "Domain name" } + required: [domain] + outputs_schema: { properties: {} } + - id: reverse_ip + name: domaintools-reverse-ip + description: "List domains hosted on an IP address." + risk: read + inputs_schema: + properties: + ip: { type: string, description: "IP address" } + required: [ip] + outputs_schema: { properties: {} } + + - id: test_connection + name: domaintools-test-connection + description: "Verify the API credentials (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/domaintools/scripts/reputation.py b/integrations/domaintools/scripts/reputation.py new file mode 100644 index 0000000..ec51fa4 --- /dev/null +++ b/integrations/domaintools/scripts/reputation.py @@ -0,0 +1,46 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://api.domaintools.com" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, cfg, params=None): + p = {k: v for k, v in (params or {}).items() if v not in (None, "")} + p["api_username"] = str(cfg.get("api_username", "")) + p["api_key"] = str(cfg.get("api_key", "")) + url = BASE + path + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="GET") + 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) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, inputs): + domain = inputs.get("domain") + if not domain: + raise Exception("domain is required") + return request("/v1/reputation", cfg, params={"domain": domain}) + + +_run(main) diff --git a/integrations/domaintools/scripts/reverse_ip.py b/integrations/domaintools/scripts/reverse_ip.py new file mode 100644 index 0000000..278f4c6 --- /dev/null +++ b/integrations/domaintools/scripts/reverse_ip.py @@ -0,0 +1,46 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://api.domaintools.com" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, cfg, params=None): + p = {k: v for k, v in (params or {}).items() if v not in (None, "")} + p["api_username"] = str(cfg.get("api_username", "")) + p["api_key"] = str(cfg.get("api_key", "")) + url = BASE + path + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="GET") + 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) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, inputs): + ip = inputs.get("ip") + if not ip: + raise Exception("ip is required") + return request("/v1/" + q(ip) + "/host-domains", cfg) + + +_run(main) diff --git a/integrations/domaintools/scripts/test_connection.py b/integrations/domaintools/scripts/test_connection.py new file mode 100644 index 0000000..b30d516 --- /dev/null +++ b/integrations/domaintools/scripts/test_connection.py @@ -0,0 +1,44 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://api.domaintools.com" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, cfg, params=None): + p = {k: v for k, v in (params or {}).items() if v not in (None, "")} + p["api_username"] = str(cfg.get("api_username", "")) + p["api_key"] = str(cfg.get("api_key", "")) + url = BASE + path + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="GET") + 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) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, inputs): + request("/v1/account", cfg) + return {"ok": True} + + +_run(main) diff --git a/integrations/domaintools/scripts/whois.py b/integrations/domaintools/scripts/whois.py new file mode 100644 index 0000000..41cb92e --- /dev/null +++ b/integrations/domaintools/scripts/whois.py @@ -0,0 +1,46 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://api.domaintools.com" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, cfg, params=None): + p = {k: v for k, v in (params or {}).items() if v not in (None, "")} + p["api_username"] = str(cfg.get("api_username", "")) + p["api_key"] = str(cfg.get("api_key", "")) + url = BASE + path + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="GET") + 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) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, inputs): + domain = inputs.get("domain") + if not domain: + raise Exception("domain is required") + return request("/v1/" + q(domain) + "/whois", cfg) + + +_run(main) diff --git a/integrations/domaintools/scripts/whois_history.py b/integrations/domaintools/scripts/whois_history.py new file mode 100644 index 0000000..ca267a0 --- /dev/null +++ b/integrations/domaintools/scripts/whois_history.py @@ -0,0 +1,46 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://api.domaintools.com" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, cfg, params=None): + p = {k: v for k, v in (params or {}).items() if v not in (None, "")} + p["api_username"] = str(cfg.get("api_username", "")) + p["api_key"] = str(cfg.get("api_key", "")) + url = BASE + path + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="GET") + 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) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, inputs): + domain = inputs.get("domain") + if not domain: + raise Exception("domain is required") + return request("/v1/" + q(domain) + "/whois/history", cfg) + + +_run(main)