fail2ban send automatic abuse email

I configured the fail2ban service to send out an automatic abuse email to the abuse contact for the IP range. To detect the right abuse contact the service is queried.
The following system tools need to be installed:

- python3
- a mailserver that also provides the sendmail binary

At first configure a new action target, a new default action and make sure that mta is set to sendmail in /etc/fail2ban/jail.conf

# ban & send an e-mail with whois report and relevant log lines to abuse contact
action_abuse = %(banaction)s[name=%(__name__)s, port="%(port)s", protocol="%(protocol)s]
               %(mta)s-abuse[name=%(__name__)s, dest="%(destemail)s", logpath=%(logpath)s]
# Choose default action.  To change, just override value of 'action' with the
# interpolation to the chosen action shortcut (e.g.  action_mw, action_mwl, etc) in jail.local
# globally (section [DEFAULT]) or per specific section
action = %(action_abuse)s

mta = sendmail

Now configure the new action_abuse in /etc/fail2ban/action.d/sendmail-abuse.conf. It's important to set the sender address on the bottom of the script!


# Option:  actionstart
# Notes.:  command executed once at the start of Fail2Ban.
# Values:  CMD
actionstart = printf %%b "Subject: [Fail2Ban] <name>: started
              Date: `date -u +"%%a, %%d %%h %%Y %%T +0000"`
              From: Fail2Ban <<sender>>
              To: <dest>\n
              The jail <name> has been started successfully.\n
              Fail2Ban" | /usr/sbin/sendmail -f <sender> <dest>

# Option:  actionstop
# Notes.:  command executed once at the end of Fail2Ban
# Values:  CMD
actionstop = printf %%b "Subject: [Fail2Ban] <name>: stopped
             Date: `date -u +"%%a, %%d %%h %%Y %%T +0000"`
             From: Fail2Ban <<sender>>
             To: <dest>\n
             The jail <name> has been stopped.\n
             Fail2Ban" | /usr/sbin/sendmail -f <sender> <dest>

# Option:  actioncheck
# Notes.:  command executed once before each actionban command
# Values:  CMD
actioncheck =

# Option:  actionban
# Notes.:  command executed when banning an IP. Take care that the
#          command is executed with Fail2Ban user rights.
# Tags:    <ip>  IP address
#          <failures>  number of failures
#          <time>  unix timestamp of the ban time
# Values:  CMD
actionban = /usr/local/sbin/fail2ban_abuse_sendmail <ip> <sender> <dest> <logpath>

# Option:  actionunban
# Notes.:  command executed when unbanning an IP. Take care that the
#          command is executed with Fail2Ban user rights.
# Tags:    <ip>  IP address
#          <failures>  number of failures
#          <time>  unix timestamp of the ban time
# Values:  CMD
actionunban =


# Defaut name of the chain
name = default

# Destination/Addressee of the mail
dest = root

# Sender of the mail
sender =

# Path to the log files which contain relevant lines for the abuser IP
logpath = /dev/null

Now the Python dependencies for the script which does the job need to be installed.

python3 -m pip install dnspython ipwhois

Create and chmod 700 the script /usr/local/sbin/fail2ban_abuse_sendmail and change the variables on the top as you wish.
If you want to have a copy of every email, add a line after 89 like msg['Bcc'] = '' or msg['Bcc'] = data['sender_mail'] in order to use the sender or msg['Bcc'] = data['dest_mail'] in order to use the recipient configured in fail2ban.

Direct download link:

#!/usr/bin/env python3
import dns.resolver
import dns.reversename
import subprocess
import sys
from email.mime.text import MIMEText
from ipwhois import IPWhois
from textwrap import dedent
import logging
from datetime import datetime
import sqlite3

# file locations
log_file = '/var/log/fail2ban_abuse_sendmail.log'
db_file = '/var/lib/fail2ban_abuse_sendmail.db'

# interval in hours for emails being send for one abuse ip. default 3 days.
retry_interval = 72

def get_abuse_contacts(reverse_ip, whois):
    d = f'{reverse_ip}'
    r = dns.resolver.query(d, 'TXT')
    # remove " from DNS TXT answer
    abuse_contacts = [i.to_text().replace('"', '') for i in r]
    # lacnic not in abusix database, replace with whois data
    if len(abuse_contacts) == 1 and '' in abuse_contacts:
      if 'emails' in whois.keys() and len(whois['emails']) >= 1:
        abuse_contacts = whois['emails']
    # build comma separated string
    return ','.join(abuse_contacts)
  except Exception as e:
    if 'emails' in whois.keys() and len(whois['emails']) >= 1:
      log('Using fallback whois recipients: ' + str(whois['emails']), 'info')
      return ','.join(whois['emails'])
      log('Error while getting Abuse contacts: ' + str(e), 'error')

def get_reverse_ip(ip):
    d = str(dns.reversename.from_address(ip))
    for i in ['', '']:
      d = d.replace(i, '')
    return d
  except Exception as e:
    log('Error while reversing IP: ' + str(e), 'error')

def whois(ip):
    obj = IPWhois(ip)
    return obj.lookup_whois()['nets'][0]
  except Exception as e:
    log('Error while querying whois: ' + str(e), 'error')

def get_logs(ip, logfile):
    logs = ''
    with open(logfile) as f:
      for line in f.readlines():
        if ip in line:
          logs += line
    return logs
  except Exception as e:
    log('Error while searching for log lines: ' + str(e), 'error')

def sendmail(data):
    # skip sending email if recipient not defined
    if data['email'] == '':
      log('Skip sending email because no recipients are defined', 'error')
    logs = get_logs(data['ip'], data['logfile'])
    whois = ''
    for k, v in data['whois'].items():
      whois += f'{k}: {v}\n'
    t = """
      This is an email abuse report about the IP address {} generated at {}
      You get this email because you are listed as the official abuse contact for this IP address.

      The following intrusion attempts were detected:

      WHOIS report:
    date_str =
    t = dedent(t).format(data['ip'], date_str, logs, whois)
    msg = MIMEText(t, 'plain', 'utf-8')
    msg['From'] = data['sender_mail']
    msg['To'] = data['email']
    msg['Subject'] = f"Automatic abuse report for IP address {data['ip']}"
    s = subprocess.Popen(['/usr/sbin/sendmail', '-t', '-oi', '-f', data['sender_mail']],
                         stdin=subprocess.PIPE, universal_newlines=True)
  except Exception as e:
    log('Error while sending email: ' + str(e), 'error')

def log(t, f):
    logger = logging.getLogger()
    fh = logging.FileHandler(log_file)
    fh.setFormatter(logging.Formatter('%(asctime)s: %(levelname)s - %(message)s'))
    for l in t.splitlines():
      if f == 'error':
      if f == 'info':
    if f == 'error':
  except Exception as e:
    sys.stderr.write('Logging error: ' + str(e))

def sql(ip, mode):
    con = sqlite3.connect(db_file, detect_types=sqlite3.PARSE_DECLTYPES)
    cur = con.cursor()
    cur.execute("""CREATE TABLE IF NOT EXISTS abusers (ip TEXT PRIMARY KEY,
                   last_mail TIMESTAMP, counter INTEGER DEFAULT 1)""")
    if mode == 'num_hours':
      cur.execute(f"SELECT last_mail FROM abusers where ip = '{ip}'")
      r = cur.fetchone()
      if r is not None:
        td = ( - r[0]).total_seconds()
        return td//3600
        return None
    if mode == 'update':
      cur.execute(f"""INSERT OR REPLACE INTO abusers (ip, last_mail, counter)
                      VALUES ('{ip}', datetime('now', 'localtime'),
                      CASE WHEN (SELECT counter FROM abusers WHERE ip = '{ip}') IS NULL THEN
                        (SELECT (counter + 1) FROM abusers WHERE ip = '{ip}')
  except Exception as e:
    log('SQLite error: ' + str(e), 'error')

def main():
    data = {}
    data['ip'] = sys.argv[1]
    data['sender_mail'] = sys.argv[2]
    data['dest_mail'] = sys.argv[3]
    data['logfile'] = sys.argv[4]
    hours_ago = sql(data['ip'], 'num_hours')
    if hours_ago is not None and hours_ago < retry_interval:
      log(f"Skip sending E-Mail for IP {data['ip']}, retrying too early ({hours_ago} hours)", 'info')
    data['whois'] = whois(data['ip'])
    data['email'] = get_abuse_contacts(get_reverse_ip(data['ip']), data['whois'])
    log(f"E-Mail sent for IP {data['ip']} with recipient(s) {data['email']}", 'info')
    sql(data['ip'], 'update')
  except Exception as e:
    log('Error while processing main: ' + str(e), 'error')

if __name__ == "__main__":
service fail2ban restart

That's it. fail2ban should now ban the source IP address and send an abuse email as well.

Schreibe einen Kommentar