#!/usr/bin/env python3 # Copyright (c) 2026 Arkadiusz Polak # Licensed under the MIT License (SPDX: MIT). # Full text: https://opensource.org/license/mit """ abuseipdb_report.py v3.3 Author: kontakt@arkadiuszpolak.pl Generates a bulk CSV of AbuseIPDB reports from LOCALLY detected CrowdSec alerts. This script does NOT send anything itself -- sending is a separate `curl` command (see README.md). ============================================================================= COMPLIANCE WITH THE AbuseIPDB REPORTING POLICY (https://www.abuseipdb.com/reporting-policy) ============================================================================= 1. "Reports of attacks older than 60 days" -- FORBIDDEN. -> Hard MAX_AGE_DAYS=60 filter on every row (we don't rely solely on --since, because --input-json may point at an old file). 2. "Report MUST contain a detailed description of the attack (port numbers, payloads, timestamps)". -> The comment includes: protocol+port, scenario names, event count, the full time window, AND real HTTP paths from the logs (payload). 3. FAQ: "limit your comments to only the key information... [avoid] email or IP address in the comment section". -> The comment does NOT contain the reported IP address, our own server's hostname, or the names of our subdomains. 4. FAQ: "for continuous abuse... report the IP roughly once per day". -> Cron runs once a day with a --since 24h window. Do NOT run this manually the same day the cron job already ran. 5. "Reports where the source address is likely spoofed (SYN/UDP floods)" -- FORBIDDEN. -> Does not apply here: every reported scenario comes from application logs (nginx/sshd), i.e. after a fully established TCP connection. 6. False reports = risk of account suspension. -> FIVE independent safeguards, see the section below. ============================================================================= SAFEGUARDS AGAINST REPORTING YOUR OWN / AN INNOCENT IP ============================================================================= 1. Filter for private/reserved/loopback addresses (internal Docker network 172.26/16, docker0 172.17/16, wg-easy VPN clients). 2. EXCLUDE_SCENARIOS -- scenarios with a documented history of false positives on the server's own legitimate traffic (http-crawl-non_statics). 3. WEAK_ONLY_SCENARIOS -- signals too weak on their own to justify a report. 4. Exclusion file -- ENABLED BY DEFAULT (~/.secrets/abuseipdb_exclude.txt), so there's no need to remember a flag in the cron job. 5. SSH auto-trust -- any IP address that has ever successfully LOGGED IN over SSH is unconditionally excluded. This is the most effective automatic defense against reporting your own, changing IP address. ============================================================================= DATA SOURCE ============================================================================= `cscli alerts list -o json`. Deliberately NOT `cscli decisions list` -- those include bans from CAPI (the community blocklist) and external lists (tor-exit-nodes, otx-webscanners), i.e. IPs that THIS server never actually observed itself. Reporting those would mean reporting someone else's detection. We additionally filter for `kind == "crowdsec"` and drop `simulated` alerts. JSON structure verified against real data from a production self-hosted server (2026-08-27). """ import argparse import csv import io import ipaddress import json import os import re import shutil import subprocess import sys from collections import defaultdict from datetime import datetime, timedelta, timezone # --- Hard limits from the AbuseIPDB documentation (bulk report) ------------- MAX_COMMENT_BYTES = 1024 # "Truncated after 1,024 characters (bytes)" MAX_ROWS = 10_000 - 1 # "less than or equal to 10,000 lines, including the headings" MAX_FILE_BYTES = 8 * 1024 * 1024 # "The CSV file must be under 8 MB" MAX_AGE_DAYS = 60 # "Timestamps MUST NOT be older than two months" # Default paths -- deliberately OUTSIDE the script's directory (i.e. outside # the git repo), next to the API key. This ensures our own IP addresses # never end up in the git history. DEFAULT_EXCLUDE_FILE = os.path.expanduser("~/.secrets/abuseipdb_exclude.txt") # ============================================================================= # !!! DO NOT TRANSLATE OR LOCALIZE THESE TEMPLATES !!! # ============================================================================= # The templates below go STRAIGHT into the public AbuseIPDB database as the # report content. AbuseIPDB is an English-language service -- the comment is # read by administrators and automated systems worldwide. # # This is the ONLY place in this script where the text sent externally is # defined. Every other message (stderr, --help) stays purely local and never # leaves the machine running the script. # # sanitize_comment() strips non-ASCII characters, but it will NOT catch # non-English text written without accented characters. The only real # guarantee is keeping these constants in plain English. # ============================================================================= TPL_SOURCE = "Detected by CrowdSec IDS on a self-hosted server." TPL_PROTO = "Target: {proto}." # protocol + port TPL_DETECTION = "Triggered rules: {scenarios}." # list of scenarios TPL_EVENTS = "{count} matching log event{plural}" # event count TPL_OBSERVED = "Observed" # when there's no event count TPL_WINDOW = "between {first} and {last} (UTC)." # time range TPL_WINDOW_SINGLE = "at {first} (UTC)." # single point in time TPL_TARGETS = "Sample requests: {targets}" # HTTP paths (payload) # Protocol/port derived from the scenario prefix -- satisfies the policy's # recommendation ("recommended port numbers") without revealing our subdomains. PROTO_BY_PREFIX = ( ("ssh-", "SSH"), ("http-", "HTTP/HTTPS (ports 80/443)"), ("nginx-", "HTTP/HTTPS (ports 80/443)"), ) PROTO_DEFAULT = "HTTP/HTTPS (ports 80/443)" # --- CrowdSec scenario -> AbuseIPDB category mapping ------------------------- # Categories per https://www.abuseipdb.com/categories: # 4=DDoS 9=Open Proxy 14=Port Scan 15=Hacking 16=SQL Injection # 18=Brute-Force 19=Bad Web Bot 20=Exploited Host 21=Web App Attack 22=SSH # Rule: NEVER assign a stronger category than what the log actually shows # (overstating severity = a false report under the policy). CATEGORY_MAP = { # --- SSH --- "ssh-bf": "18,22", "ssh-bf_user-enum": "18,22", "ssh-slow-bf": "18,22", "ssh-slow-bf_user-enum": "18,22", "ssh-time-based-bf": "18,22", "ssh-time-based-bf_user-enum": "18,22", "ssh-refused-conn": "14,22", "ssh-generic-test": "14,22", "ssh-cve-2024-6387": "15,22", # --- HTTP: scanning / recon --- "http-probing": "14,21", "http-technology-probing": "14,21", "http-generic-test": "14,21", "http-sensitive-files": "15,21", "http-path-traversal-probing": "15,21", "http-admin-interface-probing": "15,21", "http-sap-interface-probing": "15,21", "http-wordpress-scan": "19,21", "http-backdoors-attempts": "15,21", "http-bad-user-agent": "19", # --- HTTP: specific attack classes --- "http-sqli-probing": "16,21", "http-xss-probing": "15,21", "http-cve-probing": "15,21", "http-generic-bf": "18,21", "http-open-proxy": "9,21", "http-w00tw00t": "14,21", # Category 4 (DDoS) deliberately NOT used: exceeding a rate limit is not # a volumetric attack, and an inflated category counts as a false report. "nginx-req-limit-exceeded": "21", # --- Specific exploits/CVEs --- "netgear_rce": "15,21", "thinkphp-cve-2018-20062": "15,21", "grafana-cve-2021-43798": "15,21", "jira_cve-2021-26086": "15,21", } DEFAULT_CATEGORY = "15,21" # hacking + web app attack -- fallback for CVE-* entries not in the map # --- Safeguard 2: scenarios that are NEVER reported ------------------------- # `http-crawl-non_statics` triggered a false positive on the server's OWN # traffic THREE TIMES (bulk upload / WebDAV to ocis). # Reporting legitimate WebDAV traffic is exactly the kind of false report # that can get an account suspended. The scenario stays active in CrowdSec # for banning, but is NEVER included in reports. EXCLUDE_SCENARIOS = {"http-crawl-non_statics"} # --- Safeguard 3: signals too weak to justify a report on their own -------- # A merely unusual user-agent, with no attempt to access anything specific -- # typical of passive research scanners (Censys/Shodan), which by design don't # try to log in or exploit anything. An IP only gets a PASS if this is the # ONLY reason; any more specific scenario alongside it still qualifies for # a report. # Deliberately NOT done via an ASN whitelist -- an ASN only tells you where # someone rented a machine, not their intent (the same reasoning that led # this project to reject an ASN whitelist for IP bans, see RUNBOOK.md). WEAK_ONLY_SCENARIOS = {"http-bad-user-agent"} class ReportBuildError(RuntimeError): """Error that prevents safely building a report.""" def short_scenario(scenario: str) -> str: return scenario.split("/", 1)[-1] def category_for(scenario: str) -> str: return CATEGORY_MAP.get(short_scenario(scenario), DEFAULT_CATEGORY) def proto_for(scenarios) -> str: """Protocol(s) attacked by this IP. Returns ALL matches, because a single address can attempt both SSH and HTTP in the same window -- returning only the first one would give the reader information that contradicts the rule list (e.g. 'Target: HTTP' next to a triggered `ssh-bf` rule).""" labels = [] for sc in sorted(scenarios): for prefix, label in PROTO_BY_PREFIX: if sc.startswith(prefix) and label not in labels: labels.append(label) break return " + ".join(labels) if labels else PROTO_DEFAULT def truncate_bytes(text: str, limit: int) -> str: """Truncates text to `limit` BYTES (not characters) without splitting a multi-byte character. A plain text[:1024] could let a string >1024 B through.""" raw = text.encode("utf-8") if len(raw) <= limit: return text return raw[:limit].decode("utf-8", errors="ignore") # Characters that make Excel/LibreOffice interpret a cell as a formula. _FORMULA_PREFIXES = ("=", "+", "-", "@", "\t", "\r") def sanitize_comment(text: str) -> str: """Hard safeguard for comment content sent to AbuseIPDB. LANGUAGE: enforces plain ASCII -- if someone later pastes non-English text with accented characters into a template (or an attacker sends a URL containing accented characters), the CSV will only ever see the stripped, non-ASCII-free version. SECURITY: HTTP paths come FROM THE ATTACKER, so we strip control characters (which could break the CSV structure) and neutralize formula prefixes (CSV injection when the file is opened in a spreadsheet app). """ ascii_only = "".join(ch if 32 <= ord(ch) < 127 else " " for ch in text) cleaned = " ".join(ascii_only.split()) if cleaned.startswith(_FORMULA_PREFIXES): cleaned = "'" + cleaned return cleaned def load_exclusions(path): """Loads a list of IPs/CIDRs to unconditionally skip (one per line, '#' starts a comment). Add your own addresses here when you spot them in the alerts.""" nets = [] if not path: return nets try: with open(path, "r", encoding="utf-8") as f: for lineno, line in enumerate(f, 1): line = line.split("#", 1)[0].strip() if not line: continue try: nets.append(ipaddress.ip_network(line, strict=False)) except ValueError: print(f"[warning] {path}:{lineno} -- invalid entry, skipping: {line}", file=sys.stderr) except FileNotFoundError: print(f"[info] exclusion file {path} does not exist (normal if you haven't created it)", file=sys.stderr) except PermissionError: print(f"[warning] no permission to read {path} -- exclusions are NOT active!", file=sys.stderr) if nets: print(f"[info] loaded {len(nets)} exclusion entries from {path}", file=sys.stderr) return nets _IP_IN_LOG = re.compile(r"\bfrom\s+([0-9a-fA-F:.]+)\s+port\b") def harvest_ssh_trusted(days: int = MAX_AGE_DAYS): """SAFEGUARD 5 -- the most important automatic defense against reporting your own address. Collects IP addresses from which someone SUCCESSFULLY logged in over SSH in the last `days` days ("Accepted publickey/password ... from "). Such an address by definition belongs to the administrator (or someone who holds a key), not to an attacker -- it should never end up in a report. This covers the main risk scenario: a changing address from an ISP pool that was first used to work on the server, and later tripped some CrowdSec threshold. A failure to read the logs does NOT abort the run -- we return an empty list and warn loudly, since the other safeguards still apply. """ journalctl = shutil.which("journalctl") or "/usr/bin/journalctl" cmd = [journalctl, "-u", "ssh", "--since", f"-{days}d", "--grep", "Accepted", "-o", "cat", "--no-pager"] try: out = subprocess.run(cmd, capture_output=True, text=True, timeout=60).stdout except (OSError, subprocess.SubprocessError) as exc: print(f"[warning] failed to read SSH logs ({exc}) -- " f"SSH auto-trust is INACTIVE", file=sys.stderr) return [] found = set() for line in out.splitlines(): m = _IP_IN_LOG.search(line) if m: try: found.add(ipaddress.ip_network(m.group(1), strict=False)) except ValueError: continue if found: print(f"[info] SSH auto-trust: {len(found)} address(es) with a successful login " f"(last {days} days) -- these will never be reported", file=sys.stderr) else: print("[warning] SSH auto-trust: found NO successful logins at all. " "Check whether the user can read journalctl (group 'adm'/'systemd-journal'), " "otherwise this safeguard provides no protection.", file=sys.stderr) return sorted(found, key=str) def is_reportable_ip(ip_str: str, exclusions) -> tuple: """Returns (True, None) if the IP may be reported, or (False, reason). Private/loopback/reserved addresses NEVER go to AbuseIPDB -- it's a public database of internet addresses.""" try: ip = ipaddress.ip_address(ip_str) except ValueError: return False, "invalid IP address" if (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast or ip.is_unspecified): return False, "private/reserved address" for net in exclusions: if ip.version == net.version and ip in net: return False, "on the trusted/excluded list" return True, None def parse_ts(ts: str): """Parses CrowdSec's ISO8601 timestamp ('2026-08-27T07:06:43Z') into an aware datetime.""" if not ts: return None try: parsed = datetime.fromisoformat(ts.replace("Z", "+00:00")) except ValueError: return None return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) def extract_ip(alert: dict): """IP address from the alert. A 'Range' scope is deliberately NOT handled -- AbuseIPDB's bulk CSV accepts single addresses, not CIDR.""" src = alert.get("source") or {} if src.get("scope", "").lower() == "ip" and (src.get("value") or src.get("ip")): return src.get("value") or src.get("ip") val = alert.get("value", "") if val.lower().startswith("ip:"): return val.split(":", 1)[1] return None def extract_evidence(alert: dict) -> list: """Concrete evidence from events[].meta -- HTTP paths with method and response code. This is the 'payload' required by the AbuseIPDB policy. We deliberately do NOT use `target_fqdn` -- that's our own subdomains, not information about the attacker.""" out = [] for ev in alert.get("events") or []: meta = {m.get("key"): m.get("value") for m in (ev.get("meta") or [])} path = meta.get("http_path") if not path: continue verb = meta.get("http_verb", "") status = meta.get("http_status", "") out.append(f"{verb} {path}".strip() + (f" -> {status}" if status else "")) return out def fetch_alerts(since: str, limit: int): cscli = shutil.which("cscli") or "/usr/bin/cscli" # cron has a minimal PATH sudo = shutil.which("sudo") or "/usr/bin/sudo" # cscli needs to read /etc/crowdsec/local_api_credentials.yaml (root:root, 600). # The ACL granting debian:r-- on this file tends to get masked (by a chmod in # the package's postinst script, which recalculates the ACL mask) on every # CrowdSec update -- see incident on 2026-09-04 (1.7.8 -> 1.8.0). A sudo rule # with NOPASSWD (the same one used for secure.sh, /etc/sudoers.d/secure-script) # survives package updates; the ACL does not. # -n = don't prompt for a password; in a cron job with no terminal it would # just hang anyway. try: raw = subprocess.check_output( [sudo, "-n", cscli, "alerts", "list", "-o", "json", "--since", since, "--limit", str(limit)], text=True, timeout=120, ) except FileNotFoundError: raise ReportBuildError(f"could not find cscli ({cscli}) or sudo ({sudo})") except subprocess.CalledProcessError as exc: raise ReportBuildError(f"cscli exited with an error (code {exc.returncode}) - " f"check 'sudo -n cscli alerts list' manually (NOPASSWD may be missing from sudoers)") except subprocess.TimeoutExpired: raise ReportBuildError("cscli did not respond within 120 s") try: data = json.loads(raw) if raw.strip() else [] except json.JSONDecodeError as exc: raise ReportBuildError(f"cscli returned invalid JSON: {exc}") if len(data) >= limit: print(f"[warning] fetched {len(data)} alerts = the --limit ({limit}). " f"Some events may have been cut off -- consider raising --limit.", file=sys.stderr) return data def format_window(first: datetime, last: datetime) -> str: """Time window shown in the comment. When events span different DAYS, it shows the full date on both sides -- otherwise '2026-08-26T22:00:00Z..07:00:00Z' would look as if the attack ended before it started.""" fmt = "%Y-%m-%dT%H:%M:%SZ" if first == last: return TPL_WINDOW_SINGLE.format(first=first.strftime(fmt)) if first.date() == last.date(): return TPL_WINDOW.format(first=first.strftime(fmt), last=last.strftime("%H:%M:%SZ")) return TPL_WINDOW.format(first=first.strftime(fmt), last=last.strftime(fmt)) def build_rows(alerts, exclusions): grouped = defaultdict(lambda: { "cats": set(), "reasons": set(), "evidence": [], "events": 0, "first": None, "last": None, }) stats = defaultdict(int) now = datetime.now(timezone.utc) cutoff = now - timedelta(days=MAX_AGE_DAYS) for a in alerts: if a.get("kind") != "crowdsec": stats["not a local detection (kind != crowdsec)"] += 1 continue if a.get("simulated"): stats["alert in simulation mode"] += 1 continue scenario = short_scenario(a.get("scenario", "unknown")) if scenario in EXCLUDE_SCENARIOS: stats[f"scenario on the blocklist ({scenario})"] += 1 continue ip = extract_ip(a) if not ip: stats["no recognized IP (scope != Ip?)"] += 1 continue ok, reason = is_reportable_ip(ip, exclusions) if not ok: stats[f"IP rejected: {reason}"] += 1 continue ts = parse_ts(a.get("created_at")) if ts is None: stats["missing/unreadable created_at"] += 1 continue if ts < cutoff: stats[f"event older than {MAX_AGE_DAYS} days (AbuseIPDB policy)"] += 1 continue if ts > now + timedelta(minutes=5): stats["timestamp from the future (clock drift?)"] += 1 continue g = grouped[ip] g["cats"].update(c.strip() for c in category_for(scenario).split(",")) g["reasons"].add(scenario) g["evidence"].extend(extract_evidence(a)) g["events"] += a.get("events_count") or len(a.get("events") or []) if g["first"] is None or ts < g["first"]: g["first"] = ts if g["last"] is None or ts > g["last"]: g["last"] = ts rows = [] skipped_weak = [] for ip, g in grouped.items(): if g["reasons"] <= WEAK_ONLY_SCENARIOS: skipped_weak.append(ip) continue reasons = sorted(g["reasons"]) # Content built EXCLUSIVELY from the TPL_* constants (the "DO NOT # TRANSLATE" block above). No IP address, no server hostname, no # subdomains of ours. parts = [ TPL_SOURCE, TPL_PROTO.format(proto=proto_for(reasons)), TPL_DETECTION.format(scenarios=", ".join(reasons)), ] parts.append( TPL_EVENTS.format(count=g["events"], plural="s" if g["events"] != 1 else "") if g["events"] else TPL_OBSERVED ) parts.append(format_window(g["first"], g["last"])) header = " ".join(parts) + " " comment = header.rstrip() seen, samples = set(), [] for ev in g["evidence"]: if ev in seen: continue seen.add(ev) candidate = header + TPL_TARGETS.format(targets="; ".join(samples + [ev])) if len(sanitize_comment(candidate).encode("utf-8")) > MAX_COMMENT_BYTES: break samples.append(ev) comment = candidate.rstrip() final_comment = truncate_bytes(sanitize_comment(comment), MAX_COMMENT_BYTES) # Last line of defense. Deliberately NOT an assert -- assert # disappears under `python3 -O`, and this is a security invariant, # not a debug check. if not final_comment.isascii(): raise ReportBuildError(f"comment for {ip} is not pure ASCII: {final_comment!r}") rows.append([ ip, ",".join(sorted(g["cats"], key=int)), g["last"].strftime("%Y-%m-%dT%H:%M:%S+00:00"), # ISO8601 with timezone final_comment, ]) for reason, count in sorted(stats.items()): print(f"[filter] skipped {count} alert(s) -- {reason}", file=sys.stderr) if skipped_weak: print(f"[filter] skipped {len(skipped_weak)} IP(s) -- the only scenario was a weak signal " f"({', '.join(sorted(WEAK_ONLY_SCENARIOS))}): {', '.join(sorted(skipped_weak))}", file=sys.stderr) return rows def enforce_size_limits(rows): """Enforces both file limits from the AbuseIPDB documentation: max 10,000 lines (including the header) and max 8 MB.""" if len(rows) > MAX_ROWS: print(f"[warning] {len(rows)} rows > the {MAX_ROWS} limit -- truncating.", file=sys.stderr) rows = rows[:MAX_ROWS] # We measure the REAL size after CSV serialization instead of estimating -- # quoting and escaping quotation marks add bytes an estimate wouldn't see, # and underestimating would mean a file rejected by AbuseIPDB. buf = io.StringIO() writer = csv.writer(buf) writer.writerow(["IP", "Categories", "ReportDate", "Comment"]) total = len(buf.getvalue().encode("utf-8")) for i, row in enumerate(rows): buf = io.StringIO() csv.writer(buf).writerow(row) total += len(buf.getvalue().encode("utf-8")) if total > MAX_FILE_BYTES: print(f"[warning] exceeded the 8 MB limit at row {i} -- truncating the file.", file=sys.stderr) return rows[:i] return rows def write_csv(handle, rows): writer = csv.writer(handle) writer.writerow(["IP", "Categories", "ReportDate", "Comment"]) writer.writerows(rows) def main(): parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter ) parser.add_argument("--since", default="24h", help="cscli time window (default: 24h)") parser.add_argument("--limit", type=int, default=5000, help="alert limit from cscli (default: 5000)") parser.add_argument("--out", default="reports.csv", help="output CSV path") parser.add_argument("--exclude-file", default=DEFAULT_EXCLUDE_FILE, help=f"file with IPs/CIDRs to unconditionally skip " f"(default: {DEFAULT_EXCLUDE_FILE})") parser.add_argument("--no-ssh-trust", action="store_true", help="DISABLE auto-excluding addresses with a successful SSH login " "(NOT recommended -- this is the main defense against reporting your own IP)") parser.add_argument("--input-json", metavar="FILE", help="read alerts from a JSON file instead of calling cscli (test mode)") parser.add_argument("--dry-run", action="store_true", help="print the CSV to stdout instead of writing it to a file") args = parser.parse_args() try: if args.input_json: with open(args.input_json, "r", encoding="utf-8") as f: alerts = json.load(f) else: alerts = fetch_alerts(args.since, args.limit) exclusions = load_exclusions(args.exclude_file) if args.no_ssh_trust: print("[WARNING] SSH auto-trust DISABLED via --no-ssh-trust", file=sys.stderr) else: exclusions = exclusions + harvest_ssh_trusted() rows = enforce_size_limits(build_rows(alerts, exclusions)) except ReportBuildError as exc: print(f"[ERROR] {exc}", file=sys.stderr) sys.exit(2) if not rows: # A non-zero exit code breaks the `&&` chain in the cron job, so # curl will NOT send a file that contains only the header row. print("No reports qualify -- not writing a CSV file.", file=sys.stderr) sys.exit(1) if args.dry_run: write_csv(sys.stdout, rows) print(f"\n[dry-run] {len(rows)} row(s) -- nothing written, nothing sent.", file=sys.stderr) return with open(args.out, "w", newline="", encoding="utf-8") as f: write_csv(f, rows) src = f"file {args.input_json}" if args.input_json else f"the --since {args.since} window" print(f"Wrote {len(rows)} unique IP(s) to {args.out} " f"({os.path.getsize(args.out)} B, from {len(alerts)} alert(s) from {src})") if __name__ == "__main__": try: main() except BrokenPipeError: # Happens with `... --dry-run | head -5`: head closes the pipe before # we finish writing. This is not a bug in the script. On shutdown, # Python tries to flush stdout again and would print a second, # confusing traceback -- so we redirect stdout to /dev/null instead. # Pattern from the Python docs (BrokenPipeError / note on SIGPIPE). try: devnull = os.open(os.devnull, os.O_WRONLY) os.dup2(devnull, sys.stdout.fileno()) except OSError: pass sys.exit(0) except KeyboardInterrupt: print("\nInterrupted.", file=sys.stderr) sys.exit(130)