From bcc79598e6aa19d64c1034769feb0b3d686d5f11 Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sat, 11 Jul 2026 22:58:17 +0200 Subject: [PATCH] feat(ipqualityscore): new IPQualityScore enrichment integration 6 commands: IP/URL/email/phone fraud reputation, leaked-email check. API-key auth, stdlib-only. Co-Authored-By: Claude Fable 5 --- integrations/ipqualityscore/manifest.yaml | 73 +++++++++++++++++++ .../ipqualityscore/scripts/email_leaked.py | 41 +++++++++++ .../scripts/email_reputation.py | 41 +++++++++++ .../ipqualityscore/scripts/ip_reputation.py | 45 ++++++++++++ .../scripts/phone_reputation.py | 42 +++++++++++ .../ipqualityscore/scripts/test_connection.py | 41 +++++++++++ .../ipqualityscore/scripts/url_reputation.py | 41 +++++++++++ 7 files changed, 324 insertions(+) create mode 100644 integrations/ipqualityscore/manifest.yaml create mode 100644 integrations/ipqualityscore/scripts/email_leaked.py create mode 100644 integrations/ipqualityscore/scripts/email_reputation.py create mode 100644 integrations/ipqualityscore/scripts/ip_reputation.py create mode 100644 integrations/ipqualityscore/scripts/phone_reputation.py create mode 100644 integrations/ipqualityscore/scripts/test_connection.py create mode 100644 integrations/ipqualityscore/scripts/url_reputation.py diff --git a/integrations/ipqualityscore/manifest.yaml b/integrations/ipqualityscore/manifest.yaml new file mode 100644 index 0000000..0eb0f16 --- /dev/null +++ b/integrations/ipqualityscore/manifest.yaml @@ -0,0 +1,73 @@ +id: ipqualityscore +name: IPQualityScore +version: 1.0.0 +description: "IPQualityScore (API) — fraud and abuse scoring for IPs, URLs, email addresses and phone numbers (proxy/VPN/bot detection, fraud score, deliverability, leaked-data check). API-key authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: IP, URL, email and phone reputation, plus leaked-email check." +category: enrichment + +config_schema: + properties: + api_key: + type: string + description: "IPQualityScore API key" + x-soar-sensitive: true + required: + - api_key + +commands: + - id: ip_reputation + name: ipqualityscore-ip + description: "Fraud/abuse score for an IP (proxy, VPN, Tor, bot, recent abuse, fraud score)." + risk: read + inputs_schema: + properties: + ip: { type: string, description: "IP address" } + strictness: { type: number, description: "Detection strictness 0-3 (default 1)" } + required: [ip] + outputs_schema: { properties: {} } + - id: url_reputation + name: ipqualityscore-url + description: "Malware/phishing and risk scoring for a URL or domain." + risk: read + inputs_schema: + properties: + url: { type: string, description: "URL or domain" } + required: [url] + outputs_schema: { properties: {} } + - id: email_reputation + name: ipqualityscore-email + description: "Validity, deliverability and fraud/abuse scoring for an email address." + risk: read + inputs_schema: + properties: + email: { type: string, description: "Email address" } + required: [email] + outputs_schema: { properties: {} } + - id: phone_reputation + name: ipqualityscore-phone + description: "Validity, line type and fraud scoring for a phone number." + risk: read + inputs_schema: + properties: + phone: { type: string, description: "Phone number (E.164 recommended)" } + country: { type: string, description: "ISO country code hint, e.g. FR" } + required: [phone] + outputs_schema: { properties: {} } + - id: email_leaked + name: ipqualityscore-email-leaked + description: "Check whether an email address has appeared in known data breaches." + risk: read + inputs_schema: + properties: + email: { type: string, description: "Email address" } + required: [email] + outputs_schema: { properties: {} } + + - id: test_connection + name: ipqualityscore-test-connection + description: "Verify the API key (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/ipqualityscore/scripts/email_leaked.py b/integrations/ipqualityscore/scripts/email_leaked.py new file mode 100644 index 0000000..ade8cfa --- /dev/null +++ b/integrations/ipqualityscore/scripts/email_leaked.py @@ -0,0 +1,41 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://ipqualityscore.com/api/json" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(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 += "?" + 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 {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") +key = lambda: q(_cfg().get("api_key") or "") + + +def main(): + args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + email = args.get("email") + if not email: + raise Exception("email is required") + result = request("/leaked/email/" + key() + "/" + q(email)) + print(json.dumps(result)) + + +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) diff --git a/integrations/ipqualityscore/scripts/email_reputation.py b/integrations/ipqualityscore/scripts/email_reputation.py new file mode 100644 index 0000000..8144392 --- /dev/null +++ b/integrations/ipqualityscore/scripts/email_reputation.py @@ -0,0 +1,41 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://ipqualityscore.com/api/json" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(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 += "?" + 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 {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") +key = lambda: q(_cfg().get("api_key") or "") + + +def main(): + args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + email = args.get("email") + if not email: + raise Exception("email is required") + result = request("/email/" + key() + "/" + q(email)) + print(json.dumps(result)) + + +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) diff --git a/integrations/ipqualityscore/scripts/ip_reputation.py b/integrations/ipqualityscore/scripts/ip_reputation.py new file mode 100644 index 0000000..31e290c --- /dev/null +++ b/integrations/ipqualityscore/scripts/ip_reputation.py @@ -0,0 +1,45 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://ipqualityscore.com/api/json" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(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 += "?" + 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 {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") +key = lambda: q(_cfg().get("api_key") or "") + + +def main(): + args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + ip = args.get("ip") + strictness = args.get("strictness") + if not ip: + raise Exception("ip is required") + result = request( + "/ip/" + key() + "/" + q(ip), + params={"strictness": strictness if strictness not in (None, "") else 1}, + ) + print(json.dumps(result)) + + +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) diff --git a/integrations/ipqualityscore/scripts/phone_reputation.py b/integrations/ipqualityscore/scripts/phone_reputation.py new file mode 100644 index 0000000..17a2c3c --- /dev/null +++ b/integrations/ipqualityscore/scripts/phone_reputation.py @@ -0,0 +1,42 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://ipqualityscore.com/api/json" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(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 += "?" + 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 {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") +key = lambda: q(_cfg().get("api_key") or "") + + +def main(): + args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + phone = args.get("phone") + country = args.get("country") + if not phone: + raise Exception("phone is required") + result = request("/phone/" + key() + "/" + q(phone), params={"country": country or None}) + print(json.dumps(result)) + + +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) diff --git a/integrations/ipqualityscore/scripts/test_connection.py b/integrations/ipqualityscore/scripts/test_connection.py new file mode 100644 index 0000000..ebdb489 --- /dev/null +++ b/integrations/ipqualityscore/scripts/test_connection.py @@ -0,0 +1,41 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://ipqualityscore.com/api/json" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(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 += "?" + 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 {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") +key = lambda: q(_cfg().get("api_key") or "") + + +def main(): + result = request("/ip/" + key() + "/8.8.8.8") + if not isinstance(result, dict) or "success" not in result: + raise Exception("unexpected response") + if result.get("success") is False: + raise Exception(result.get("message") or "authentication failed") + 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) diff --git a/integrations/ipqualityscore/scripts/url_reputation.py b/integrations/ipqualityscore/scripts/url_reputation.py new file mode 100644 index 0000000..253bd46 --- /dev/null +++ b/integrations/ipqualityscore/scripts/url_reputation.py @@ -0,0 +1,41 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +API = "https://ipqualityscore.com/api/json" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def request(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 += "?" + 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 {} + + +q = lambda v: urllib.parse.quote(str(v), safe="") +key = lambda: q(_cfg().get("api_key") or "") + + +def main(): + args = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + url = args.get("url") + if not url: + raise Exception("url is required") + result = request("/url/" + key() + "/" + q(url)) + print(json.dumps(result)) + + +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)