From 9504b22e04aa6d6082adb9e0824dba20288dba9f Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sat, 11 Jul 2026 22:33:03 +0200 Subject: [PATCH] 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 --- integrations/abuseipdb/manifest.yaml | 67 +++++++++++++++++++ integrations/abuseipdb/scripts/check_block.py | 45 +++++++++++++ integrations/abuseipdb/scripts/check_ip.py | 48 +++++++++++++ .../abuseipdb/scripts/get_blacklist.py | 46 +++++++++++++ integrations/abuseipdb/scripts/report_ip.py | 50 ++++++++++++++ .../abuseipdb/scripts/test_connection.py | 40 +++++++++++ 6 files changed, 296 insertions(+) create mode 100644 integrations/abuseipdb/manifest.yaml create mode 100644 integrations/abuseipdb/scripts/check_block.py create mode 100644 integrations/abuseipdb/scripts/check_ip.py create mode 100644 integrations/abuseipdb/scripts/get_blacklist.py create mode 100644 integrations/abuseipdb/scripts/report_ip.py create mode 100644 integrations/abuseipdb/scripts/test_connection.py diff --git a/integrations/abuseipdb/manifest.yaml b/integrations/abuseipdb/manifest.yaml new file mode 100644 index 0000000..5fb6a13 --- /dev/null +++ b/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: {} } diff --git a/integrations/abuseipdb/scripts/check_block.py b/integrations/abuseipdb/scripts/check_block.py new file mode 100644 index 0000000..a8942e2 --- /dev/null +++ b/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) diff --git a/integrations/abuseipdb/scripts/check_ip.py b/integrations/abuseipdb/scripts/check_ip.py new file mode 100644 index 0000000..14499bb --- /dev/null +++ b/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) diff --git a/integrations/abuseipdb/scripts/get_blacklist.py b/integrations/abuseipdb/scripts/get_blacklist.py new file mode 100644 index 0000000..cd8d640 --- /dev/null +++ b/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) diff --git a/integrations/abuseipdb/scripts/report_ip.py b/integrations/abuseipdb/scripts/report_ip.py new file mode 100644 index 0000000..ea76a19 --- /dev/null +++ b/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) diff --git a/integrations/abuseipdb/scripts/test_connection.py b/integrations/abuseipdb/scripts/test_connection.py new file mode 100644 index 0000000..cb0fe5e --- /dev/null +++ b/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)