feat(abuseipdb): new AbuseIPDB enrichment integration

5 commands: IP abuse-reputation check, report abusive IP, blacklist
retrieval, CIDR-block check. API-key auth, stdlib-only.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

integrations/abuseipdb/manifest.yaml

@@ -0,0 +1,67 @@
+id: abuseipdb
+name: AbuseIPDB
+version: 1.0.0
+description: "AbuseIPDB (API v2) — check the abuse reputation of an IP, report abusive IPs, pull the blacklist and check a CIDR block. API-key authentication; stdlib-only, no extra Python dependencies."
+changelog: "1.0.0 — Initial release: IP reputation check, report, blacklist retrieval and CIDR-block check."
+category: enrichment
+
+config_schema:
+  properties:
+    api_key:
+      type: string
+      description: "AbuseIPDB API key"
+      x-soar-sensitive: true
+  required:
+    - api_key
+
+commands:
+  - id: check_ip
+    name: abuseipdb-check-ip
+    description: "Check the abuse-confidence reputation of an IP address."
+    risk: read
+    inputs_schema:
+      properties:
+        ip: { type: string, description: "IP address to check" }
+        max_age_days: { type: number, description: "Only consider reports within this many days (default 30, max 365)" }
+        verbose: { type: boolean, description: "Include the detailed report list" }
+      required: [ip]
+    outputs_schema: { properties: {} }
+  - id: report_ip
+    name: abuseipdb-report-ip
+    description: "Report an abusive IP address to AbuseIPDB."
+    inputs_schema:
+      properties:
+        ip: { type: string, description: "IP address to report" }
+        categories: { type: string, description: "Comma-separated AbuseIPDB category IDs (e.g. 18,22)" }
+        comment: { type: string, description: "Description of the abusive activity (avoid sensitive data)" }
+      required: [ip, categories]
+    outputs_schema: { properties: {} }
+  - id: get_blacklist
+    name: abuseipdb-get-blacklist
+    description: "Retrieve the AbuseIPDB blacklist of the most-reported IPs."
+    risk: read
+    inputs_schema:
+      properties:
+        confidence_minimum: { type: number, description: "Minimum abuse-confidence score (default 100)" }
+        limit: { type: number, description: "Maximum entries (default 100)" }
+      required: []
+    outputs_schema: { properties: {} }
+  - id: check_block
+    name: abuseipdb-check-block
+    description: "Check the reports for every address in a CIDR block (max /24 on the free tier)."
+    risk: read
+    inputs_schema:
+      properties:
+        network: { type: string, description: "CIDR network, e.g. 192.0.2.0/24" }
+        max_age_days: { type: number, description: "Only consider reports within this many days (default 30)" }
+      required: [network]
+    outputs_schema: { properties: {} }
+
+  - id: test_connection
+    name: abuseipdb-test-connection
+    description: "Verify the API key (used by the Test button)."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }

integrations/abuseipdb/scripts/check_block.py

@@ -0,0 +1,45 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.abuseipdb.com/api/v2"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None, body=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    data = urllib.parse.urlencode(body).encode("utf-8") if body is not None else None
+    headers = {"Accept": "application/json", "Key": str(_cfg().get("api_key") or "")}
+    if data is not None:
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+    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 main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    network = inputs.get("network")
+    if not network:
+        raise Exception("network is required")
+    max_age_days = inputs.get("max_age_days")
+
+    params = {"network": network, "maxAgeInDays": max_age_days or 30}
+
+    result = request("GET", "/check-block", params=params)
+    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/abuseipdb/scripts/check_ip.py

@@ -0,0 +1,48 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.abuseipdb.com/api/v2"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None, body=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    data = urllib.parse.urlencode(body).encode("utf-8") if body is not None else None
+    headers = {"Accept": "application/json", "Key": str(_cfg().get("api_key") or "")}
+    if data is not None:
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+    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 main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    ip = inputs.get("ip")
+    if not ip:
+        raise Exception("ip is required")
+    max_age_days = inputs.get("max_age_days")
+    verbose = inputs.get("verbose")
+
+    params = {"ipAddress": ip, "maxAgeInDays": max_age_days or 30}
+    if verbose:
+        params["verbose"] = ""
+
+    result = request("GET", "/check", params=params)
+    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/abuseipdb/scripts/get_blacklist.py

@@ -0,0 +1,46 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.abuseipdb.com/api/v2"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None, body=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    data = urllib.parse.urlencode(body).encode("utf-8") if body is not None else None
+    headers = {"Accept": "application/json", "Key": str(_cfg().get("api_key") or "")}
+    if data is not None:
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+    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 main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    confidence_minimum = inputs.get("confidence_minimum")
+    limit = inputs.get("limit")
+
+    params = {
+        "confidenceMinimum": confidence_minimum or 100,
+        "limit": limit or 100,
+    }
+
+    result = request("GET", "/blacklist", params=params)
+    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/abuseipdb/scripts/report_ip.py

@@ -0,0 +1,50 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.abuseipdb.com/api/v2"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None, body=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    data = urllib.parse.urlencode(body).encode("utf-8") if body is not None else None
+    headers = {"Accept": "application/json", "Key": str(_cfg().get("api_key") or "")}
+    if data is not None:
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+    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 main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    ip = inputs.get("ip")
+    if not ip:
+        raise Exception("ip is required")
+    categories = inputs.get("categories")
+    if not categories:
+        raise Exception("categories is required")
+    comment = inputs.get("comment")
+
+    body = {"ip": ip, "categories": categories}
+    if comment:
+        body["comment"] = comment
+
+    result = request("POST", "/report", body=body)
+    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/abuseipdb/scripts/test_connection.py

@@ -0,0 +1,40 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+API = "https://api.abuseipdb.com/api/v2"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def request(method, path, params=None, body=None):
+    url = API + path
+    q = {k: str(x) for k, x in (params or {}).items() if x not in (None, "")}
+    if q:
+        url += ("&" if "?" in url else "?") + urllib.parse.urlencode(q)
+    data = urllib.parse.urlencode(body).encode("utf-8") if body is not None else None
+    headers = {"Accept": "application/json", "Key": str(_cfg().get("api_key") or "")}
+    if data is not None:
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+    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 main():
+    params = {"ipAddress": "8.8.8.8", "maxAgeInDays": 1}
+    result = request("GET", "/check", params=params)
+    if "data" not in result:
+        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)