Fail2Ban, CSF и Suricata

install -m 600 /dev/null /etc/fail2ban/abuseipdb.key
printf '%s\n' 'YOUR_OWN_API_KEY' > /etc/fail2ban/abuseipdb.key
chmod 600 /etc/fail2ban/abuseipdb.key

[Definition]

# Отправляет жалобу в AbuseIPDB.RU при бане.
# Отличие от стокового abuseipdb.conf — только домен api.abuseipdb.ru.
actionban = curl --fail --silent --show-error --max-time 10 \
    'https://api.abuseipdb.ru/api/v2/report' \
    -H 'Accept: application/json' \
    -H 'Key: <abuseipdb_apikey>' \
    --data-urlencode 'comment=<matches>' \
    --data-urlencode 'ip=<ip>' \
    --data 'categories=<abuseipdb_category>'

[Init]
# Категории через запятую, ID 1..23. 18 = Brute-Force, 22 = SSH.
abuseipdb_category = 18,22

# Ключ читается из файла, чтобы не хранить его в открытом виде в jail.local.
# Если файла нет — подставьте ключ напрямую (и ограничьте права на конфиг).
abuseipdb_apikey = %(abuseipdb_apikey)s

[Definition]
actionban = /usr/local/bin/abuseipdb-report '<ip>' '<abuseipdb_category>' '<matches>'

[Init]
abuseipdb_category = 18,22

#!/usr/bin/env bash
# /usr/local/bin/abuseipdb-report — отправка одной жалобы из Fail2Ban.
set -euo pipefail

IP="$1"
CATEGORIES="${2:-18,22}"
MATCHES="${3:-banned by fail2ban}"
KEY="$(cat /etc/fail2ban/abuseipdb.key)"

curl --fail --silent --show-error --max-time 10 \
  'https://api.abuseipdb.ru/api/v2/report' \
  -H 'Accept: application/json' \
  -H "Key: ${KEY}" \
  --data-urlencode "ip=${IP}" \
  --data-urlencode "comment=${MATCHES}" \
  --data "categories=${CATEGORIES}"

chmod 700 /usr/local/bin/abuseipdb-report

[sshd]
enabled  = true
port     = ssh
filter   = sshd
logpath  = /var/log/auth.log
maxretry = 5
findtime = 10m
bantime  = 1h

# важное: action = обычный бан + отправка жалобы
action = iptables-multiport[name=sshd, port="ssh", protocol=tcp]
         abuseipdb.ru

fail2ban-client -t                     # проверка конфигурации
systemctl restart fail2ban
fail2ban-client status sshd            # статус jail и список забаненных

# Ручной вызов action — отправит жалобу на публичный адрес (подставьте свой или адрес нарушителя)
abuseipdb-report 89.205.125.160 18,22 'manual test from fail2ban'

# Через API убеждаемся, что жалоба видна (ключ тот же)
curl -sG https://api.abuseipdb.ru/api/v2/check \
  --data-urlencode "ipAddress=127.0.0.1" \
  -d maxAgeInDays=1 -d verbose \
  -H "Key: $(cat /etc/fail2ban/abuseipdb.key)" \
  -H "Accept: application/json"

journalctl -u fail2ban -n 50 --no-pager
grep -i abuseipdb /var/log/fail2ban.log | tail

# 1. Проверить, что фильтр ловит строки нужного формата
fail2ban-regex /var/log/auth.log /etc/fail2ban/filter.d/sshd.conf

# 2. Проверить итоговое действие (dry-run без сети)
fail2ban-client get sshd actions

#!/usr/bin/env bash
# /usr/local/bin/csf-abuseipdb-sync — обновление csf.deny из AbuseIPDB.RU
set -euo pipefail

KEY="$(cat /etc/fail2ban/abuseipdb.key)"
OUT=/tmp/abuseipdb.csf.deny
TMP="$(mktemp)"

curl -sS -G https://api.abuseipdb.ru/api/v2/blacklist \
  -d confidenceMinimum=75 \
  -d limit=10000 \
  -d plaintext \
  -H "Key: ${KEY}" \
  -H "Accept: text/plain" \
  -D /tmp/abuseipdb.headers \
  -o "$TMP"

# Проверяем, что ответ непустой
test -s "$TMP" || { rm -f "$TMP"; exit 1; }

# Формируем csf.deny: # BEGIN/END-маркеры, чтобы не удалять чужие правила
{
  echo '# BEGIN AbuseIPDB.RU (auto-managed, do not edit)'
  grep -vE '^\s*$|^\s*#' "$TMP" | awk '{print $1 " # AbuseIPDB.RU confidence>=75"}'
  echo '# END AbuseIPDB.RU'
} > "$OUT"

# Заменяем только наш блок в csf.deny
python3 - "$OUT" <<'PY'
import re
import sys

new_block = open(sys.argv[1], encoding="utf-8").read()
path = "/etc/csf/csf.deny"
try:
    old = open(path, encoding="utf-8").read()
except FileNotFoundError:
    old = ""
pattern = re.compile(r"# BEGIN AbuseIPDB\.RU.*?# END AbuseIPDB\.RU\n?", re.S)
old = pattern.sub("", old).rstrip() + "\n\n"
open(path, "w", encoding="utf-8").write(old + new_block)
PY

rm -f "$TMP"
csf -r >/dev/null 2>&1 || true

*/30 * * * * /usr/local/bin/csf-abuseipdb-sync >> /var/log/csf-abuseipdb.log 2>&1

#!/usr/bin/env python3
"""suricata-abuseipdb.py — жалобы из EVE JSON Suricata в AbuseIPDB.RU."""
import json
import os
import sqlite3
import time
import urllib.parse
import urllib.request

EVE = "/var/log/suricata/eve.json"
STATE = "/var/lib/suricata-abuseipdb.sqlite"
API_KEY = open("/etc/fail2ban/abuseipdb.key", encoding="utf-8").read().strip()
BASE = "https://api.abuseipdb.ru/api/v2"
# Категории: 14 = Port Scan, 21 = Web App Attack — подберите под свои правила
CATEGORIES = "14,21"

db = sqlite3.connect(STATE)
db.execute("CREATE TABLE IF NOT EXISTS sent (ip TEXT PRIMARY KEY, sent_at INTEGER)")
db.commit()

now = int(time.time())
sent = 0
with open(EVE, "r", encoding="utf-8", errors="replace") as handle:
    for line in handle:
        try:
            event = json.loads(line)
        except ValueError:
            continue
        if event.get("event_type") != "alert" or event.get("alert", {}).get("severity", 3) > 2:
            continue
        ip = event.get("src_ip", "")
        if not ip:
            continue
        row = db.execute("SELECT sent_at FROM sent WHERE ip = ?", (ip,)).fetchone()
        if row and now - row[0] < 3600:  # не чаще раза в час на адрес
            continue
        payload = urllib.parse.urlencode(
            {"ip": ip, "categories": CATEGORIES, "comment": event["alert"].get("signature", "Suricata alert")[:1024]}
        ).encode()
        request = urllib.request.Request(
            f"{BASE}/report",
            data=payload,
            headers={"Key": API_KEY, "Accept": "application/json", "Content-Type": "application/x-www-form-urlencoded"},
            method="POST",
        )
        try:
            with urllib.request.urlopen(request, timeout=10) as response:
                if response.status == 200:
                    sent += 1
        except Exception as error:  # 429 дедупликации — не ошибка, пропускаем
            print("report failed:", ip, error)
        db.execute("INSERT OR REPLACE INTO sent (ip, sent_at) VALUES (?, ?)", (ip, now))
        db.commit()

print("sent:", sent)

*/10 * * * * /usr/local/bin/suricata-abuseipdb.py >> /var/log/suricata-abuseipdb.log 2>&1

import requests

API_KEY = "YOUR_OWN_API_KEY"
_cache: dict[str, int] = {}

def score(ip: str) -> int:
    if ip in _cache:
        return _cache[ip]
    response = requests.get(
        "https://api.abuseipdb.ru/api/v2/check",
        params={"ipAddress": ip},
        headers={"Key": API_KEY, "Accept": "application/json"},
        timeout=10,
    )
    response.raise_for_status()
    _cache[ip] = response.json()["data"]["abuseConfidenceScore"]
    return _cache[ip]