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 <noreply@anthropic.com>

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: {} }

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)

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)

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)

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)

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)

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)