sjlongland/fail2ban-subnet-scrape.py

## fail2ban-subnet-scrape.py
#!/usr/bin/env python3

"""
Automatically coalesce IPs from fail2ban into subnets and flag them for
banning.
"""

# © 2023 Stuart Longland <me@vk4msl.com>
# SPDX-License-Identifier: BSD-2-Clause

# Usage, run this periodically from `cron`, then configure your
# `fail2ban` instance to actually act on the bans:
#
# ==> /etc/fail2ban/filter.d/fail2ban-subnet.conf <==
# [INCLUDES]
#
# before = common.conf
#
# [Definition]
#
# failregex = fail2ban-subnet-scrape\.py: <SUBNET> is currently misbehaving$
#
# ==> /etc/fail2ban/jail.d/subnet.conf <==
# [subnet]
# enabled  = true
# filter   = fail2ban-subnet
# banaction= iptables-allports
# logpath  = /var/log/messages
# maxretry = 1
#
# Known bugs:
# - scraping /var/log/fail2ban.log is horrible, it'd be nice to retrieve all
#   the "recently seen" IPs from the `fail2ban-client` (or better yet, import
#   `fail2ban`'s API since it is Python too), but it works.
# - it'd also be nice to "filter out" the subnets currently banned so we can
#   avoid the "WARNING: <subnet> is already banned"


import re
import sys
import os
import time
import datetime
import ipaddress
import sqlite3
import hashlib
import logging
import syslog

# fail2ban.log full path
FAIL2BAN_LOG = "/var/log/fail2ban.log"

# Subnet status database
SUBNET_DB = "/var/lib/fail2ban/subnet.sqlite3"

# Min subnet length: we don't block anything bigger than this
# e.g. 16 subnet bits minimum → do not block subnets bigger than a /16
# One for IPv4; the other for IPv6
MIN_SUBNET4_LENGTH = 16
MIN_SUBNET6_LENGTH = 56

# Max subnet length: we don't block anything smaller than this
# One for IPv4; the other for IPv6
MAX_SUBNET4_LENGTH = 28
MAX_SUBNET6_LENGTH = 64

# Ignore these jails
IGNORE_JAILS = set(["subnet"])

# Step size for subnet length
SUBNET_LENGTH_STEP = 4

# Base number of strikes before we ban a subnet
SUBNET_STRIKES = 4

# Adjustment based on subnet size, bigger subnets require more hits
SUBNET_STRIKES_SIZE_FACTOR = 2

# Decrement delay: after N seconds, we will decrement a record that has
# not been seen a second time.
DECREMENT_DELAY = 14400

# How long do we ban whole subnets for?
SUBNET_BAN_EXPIRY = 86400

# fail2ban.log date/time format (we'll ignore milliseconds)
# Assume system local time
LOG_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S"

# fail2ban.log log format regular expression
LOG_FORMAT_RE = re.compile(
    r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3} "  # log timestamp
    r"fail2ban\.filter +\[\d+\]: +INFO +"  # logger name and level
    # jail name, offending address, daemon log date/time
    r"\[([^\[\]]+)\] Found (.+) - (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})$"
)

# Initialise logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("fail2ban.subnet")

# Initialise or open the database
db_new = not os.path.exists(SUBNET_DB)
db = sqlite3.connect(SUBNET_DB)

if db_new:
    # Create the tables
    db.execute(
        """
        CREATE TABLE lastrun (
            path TEXT NOT NULL,
            firstlinemd5 TEXT,
            lastfilepos INTEGER,
            mtime INTEGER
        );
    """
    )
    db.execute(
        """
        CREATE UNIQUE INDEX lastrun_path ON lastrun(path);
    """
    )
    db.execute(
        """
        CREATE TABLE found_ips (
            ip TEXT NOT NULL,
            count INTEGER NOT NULL DEFAULT 0,
            lastseen INTEGER NOT NULL,
            decrement INTEGER NOT NULL
        );
    """
    )
    db.execute(
        """
        CREATE UNIQUE INDEX found_ips_ip ON found_ips(ip);
    """
    )

# Pick up the last-known state
logfile_firstlinemd5 = None
logfile_lastfilepos = None
logfile_mtime = 0

for (logfile_firstlinemd5, logfile_filelinepos, logfile_mtime) in db.execute(
    """
    SELECT firstlinemd5, lastfilepos, mtime
    FROM lastrun
    WHERE path=?
    """,
    (FAIL2BAN_LOG,),
):
    # Should only be one record
    break

# Get the current log file mtime
logfile_mtime_now = os.stat(FAIL2BAN_LOG).st_mtime
if logfile_mtime_now <= logfile_mtime:
    # Nothing changed
    logger.info("Log file has not changed")
    sys.exit(0)

# Scan the log file
ips = {}
for (linenum, line) in enumerate(open(FAIL2BAN_LOG, "r")):
    if linenum == 0:
        # First line, does it match the MD5 of our existing line
        md5hash = hashlib.md5()
        md5hash.update(line.encode())
        md5 = md5hash.hexdigest()
        logger.debug(
            "First line MD5: was %s now %s", logfile_firstlinemd5, md5
        )

        if md5 != logfile_firstlinemd5:
            # MD5 has changed, start at the beginning!
            logger.info(
                "Log file newly rolled over, starting at the beginning."
            )
            logfile_filelinepos = 0
            logfile_firstlinemd5 = md5
        else:
            logger.info(
                "Log file continues, seeking to position %d.",
                logfile_filelinepos,
            )

    if linenum < logfile_filelinepos:
        # We've seen this, carry on!
        continue

    match = LOG_FORMAT_RE.match(line)
    if not match:
        logger.debug("Not matching: %r", line)
        continue

    try:
        (jail, ip_str, ts_str) = match.groups()
        if jail in IGNORE_JAILS:
            continue

        ts = datetime.datetime.strptime(ts_str, LOG_TIMESTAMP_FORMAT)

        ips[ipaddress.ip_address(ip_str)] = int(ts.timestamp())
    except ValueError:
        # Unparseable data, continue
        logger.debug("Not parsed: %r", line, exc_info=1)
        continue

if logfile_firstlinemd5 is None:
    # File is empty
    sys.exit(0)

# Record our position
logfile_lastfilepos = linenum

# Make a note of the IPs we just saw
NOW = int(time.time())
for (ip, ts) in ips.items():
    # Did we have a count for this already?
    count = 0
    for (count,) in db.execute(
        "SELECT count FROM found_ips WHERE ip=?;", (str(ip),)
    ):
        # Should be just one record
        break

    # New values
    count += 1
    decrement = ts + DECREMENT_DELAY

    db.execute(
        """
        INSERT INTO found_ips
            (ip, count, lastseen, decrement)
        VALUES
            (?,  ?,     ?,        ?)
        ON CONFLICT (ip)
        DO UPDATE SET
            count=?, lastseen=?, decrement=?;
        """,
        (str(ip), count, ts, decrement, count, ts, decrement),
    )
    logger.info(
        "%s seen %d time(s), last at %s, decrement at %s",
        ip,
        count,
        ts,
        decrement,
    )

# Update the last-seen record
db.execute(
    """
    INSERT INTO lastrun
        (path, firstlinemd5, lastfilepos, mtime)
    VALUES
        (?,    ?,            ?,           ?)
    ON CONFLICT (path)
    DO UPDATE SET
        firstlinemd5=?, lastfilepos=?, mtime=?
    """,
    (
        FAIL2BAN_LOG,
        logfile_firstlinemd5,
        logfile_lastfilepos,
        logfile_mtime_now,
        logfile_firstlinemd5,
        logfile_lastfilepos,
        logfile_mtime_now,
    ),
)
logger.info(
    "recording file %r firstline=%s, pos=%d, mtime=%s",
    FAIL2BAN_LOG,
    logfile_firstlinemd5,
    logfile_lastfilepos,
    logfile_mtime_now,
)

db.commit()

# Decrement those who are behaving
db.execute("UPDATE found_ips SET count=count-1 WHERE decrement < ?;", (NOW,))

# Purge those who have continued behaving
db.execute("DELETE FROM found_ips WHERE count <= 0;")
db.commit()

# Coalesce misbehaving subnets
subnets = {}
for (ip_str, count, lastseen) in db.execute(
    """
    SELECT ip, count, lastseen FROM found_ips;
"""
):
    ip = ipaddress.ip_address(ip_str)

    # IPv4 or IPv6?
    if ip.version == 4:
        min_length = MIN_SUBNET4_LENGTH
        max_length = MAX_SUBNET4_LENGTH
    else:
        min_length = MIN_SUBNET6_LENGTH
        max_length = MAX_SUBNET6_LENGTH

    for length in range(
        min_length, max_length + SUBNET_LENGTH_STEP, SUBNET_LENGTH_STEP
    ):
        # strict=False → ignore host bits
        subnet = ipaddress.ip_network((ip, length), strict=False)

        try:
            (subnet_count, subnet_lastseen) = subnets[subnet]
        except KeyError:
            subnet_count = 0
            subnet_lastseen = 0

        subnet_count += count
        subnet_lastseen = max(subnet_lastseen, lastseen)
        logger.debug(
            "%s seen %d time(s) last at %s",
            subnet,
            subnet_count,
            subnet_lastseen,
        )
        subnets[subnet] = (subnet_count, subnet_lastseen)

# Figure out who of these are misbehaving
subnet_expiry = NOW - SUBNET_BAN_EXPIRY
misbehaving = set()
for (subnet, (count, lastseen)) in subnets.items():
    # IPv4 or IPv6?
    if ip.version == 4:
        max_length = MAX_SUBNET4_LENGTH
    else:
        max_length = MAX_SUBNET6_LENGTH

    # Size factor for the subnet
    size = round((max_length - subnet.prefixlen) / SUBNET_LENGTH_STEP)
    logger.debug("%s is size %d", subnet, size)

    # Compute number of strikes
    strikes = SUBNET_STRIKES + round(pow(SUBNET_STRIKES_SIZE_FACTOR, size))
    logger.debug("will accept %d strikes, has %d strikes", strikes, count)

    if (count >= strikes) and (lastseen > subnet_expiry):
        # Still misbehaving, don't bother looking at smaller subnets
        logger.info(
            "%s is misbehaving, seen %d time(s), last at %s",
            subnet,
            count,
            lastseen,
        )
        misbehaving.add(subnet)

# Reduce overlapping subnets to the smallest possible set
done = False
while not done:
    done = True
    logger.debug(
        "Cleaning up overlapping subnets: %d subnet(s)", len(misbehaving)
    )
    for subnet in list(misbehaving):
        children = [
            s
            for s in misbehaving
            if (subnet is not s) and subnet.supernet_of(s)
        ]
        num_children = len(children)
        if num_children == 1:
            # Keep the child, ditch the parent
            logger.debug("Keeping %s in favour of %s", children[0], subnet)
            misbehaving.discard(subnet)
            done = False
            break
        elif num_children > 1:
            # Prefer the parent since it's clear _multiple_ addresses in this
            # subnet are behaving badly.
            logger.debug(
                "Keeping %s in favour of %s",
                subnet,
                ", ".join([str(s) for s in children]),
            )
            for s in children:
                misbehaving.discard(s)
            done = False
            break


if misbehaving:
    logger.info(
        "All misbehaving subnets: %s",
        ", ".join([str(s) for s in misbehaving]),
    )

    for subnet in misbehaving:
        syslog.syslog(
            syslog.LOG_NOTICE | syslog.LOG_DAEMON,
            "%s is currently misbehaving" % subnet,
        )
	#!/usr/bin/env python3

	"""
	Automatically coalesce IPs from fail2ban into subnets and flag them for
	banning.
	"""

	# © 2023 Stuart Longland <me@vk4msl.com>
	# SPDX-License-Identifier: BSD-2-Clause

	# Usage, run this periodically from `cron`, then configure your
	# `fail2ban` instance to actually act on the bans:
	#
	# ==> /etc/fail2ban/filter.d/fail2ban-subnet.conf <==
	# [INCLUDES]
	#
	# before = common.conf
	#
	# [Definition]
	#
	# failregex = fail2ban-subnet-scrape\.py: <SUBNET> is currently misbehaving$
	#
	# ==> /etc/fail2ban/jail.d/subnet.conf <==
	# [subnet]
	# enabled = true
	# filter = fail2ban-subnet
	# banaction= iptables-allports
	# logpath = /var/log/messages
	# maxretry = 1
	#
	# Known bugs:
	# - scraping /var/log/fail2ban.log is horrible, it'd be nice to retrieve all
	# the "recently seen" IPs from the `fail2ban-client` (or better yet, import
	# `fail2ban`'s API since it is Python too), but it works.
	# - it'd also be nice to "filter out" the subnets currently banned so we can
	# avoid the "WARNING: <subnet> is already banned"


	import re
	import sys
	import os
	import time
	import datetime
	import ipaddress
	import sqlite3
	import hashlib
	import logging
	import syslog

	# fail2ban.log full path
	FAIL2BAN_LOG = "/var/log/fail2ban.log"

	# Subnet status database
	SUBNET_DB = "/var/lib/fail2ban/subnet.sqlite3"

	# Min subnet length: we don't block anything bigger than this
	# e.g. 16 subnet bits minimum → do not block subnets bigger than a /16
	# One for IPv4; the other for IPv6
	MIN_SUBNET4_LENGTH = 16
	MIN_SUBNET6_LENGTH = 56

	# Max subnet length: we don't block anything smaller than this
	# One for IPv4; the other for IPv6
	MAX_SUBNET4_LENGTH = 28
	MAX_SUBNET6_LENGTH = 64

	# Ignore these jails
	IGNORE_JAILS = set(["subnet"])

	# Step size for subnet length
	SUBNET_LENGTH_STEP = 4

	# Base number of strikes before we ban a subnet
	SUBNET_STRIKES = 4

	# Adjustment based on subnet size, bigger subnets require more hits
	SUBNET_STRIKES_SIZE_FACTOR = 2

	# Decrement delay: after N seconds, we will decrement a record that has
	# not been seen a second time.
	DECREMENT_DELAY = 14400

	# How long do we ban whole subnets for?
	SUBNET_BAN_EXPIRY = 86400

	# fail2ban.log date/time format (we'll ignore milliseconds)
	# Assume system local time
	LOG_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S"

	# fail2ban.log log format regular expression
	LOG_FORMAT_RE = re.compile(
	r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3} " # log timestamp
	r"fail2ban\.filter +\[\d+\]: +INFO +" # logger name and level
	# jail name, offending address, daemon log date/time
	r"\[([^\[\]]+)\] Found (.+) - (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})$"
	)

	# Initialise logging
	logging.basicConfig(level=logging.DEBUG)
	logger = logging.getLogger("fail2ban.subnet")

	# Initialise or open the database
	db_new = not os.path.exists(SUBNET_DB)
	db = sqlite3.connect(SUBNET_DB)

	if db_new:
	# Create the tables
	db.execute(
	"""
	CREATE TABLE lastrun (
	path TEXT NOT NULL,
	firstlinemd5 TEXT,
	lastfilepos INTEGER,
	mtime INTEGER
	);
	"""
	)
	db.execute(
	"""
	CREATE UNIQUE INDEX lastrun_path ON lastrun(path);
	"""
	)
	db.execute(
	"""
	CREATE TABLE found_ips (
	ip TEXT NOT NULL,
	count INTEGER NOT NULL DEFAULT 0,
	lastseen INTEGER NOT NULL,
	decrement INTEGER NOT NULL
	);
	"""
	)
	db.execute(
	"""
	CREATE UNIQUE INDEX found_ips_ip ON found_ips(ip);
	"""
	)

	# Pick up the last-known state
	logfile_firstlinemd5 = None
	logfile_lastfilepos = None
	logfile_mtime = 0

	for (logfile_firstlinemd5, logfile_filelinepos, logfile_mtime) in db.execute(
	"""
	SELECT firstlinemd5, lastfilepos, mtime
	FROM lastrun
	WHERE path=?
	""",
	(FAIL2BAN_LOG,),
	):
	# Should only be one record
	break

	# Get the current log file mtime
	logfile_mtime_now = os.stat(FAIL2BAN_LOG).st_mtime
	if logfile_mtime_now <= logfile_mtime:
	# Nothing changed
	logger.info("Log file has not changed")
	sys.exit(0)

	# Scan the log file
	ips = {}
	for (linenum, line) in enumerate(open(FAIL2BAN_LOG, "r")):
	if linenum == 0:
	# First line, does it match the MD5 of our existing line
	md5hash = hashlib.md5()
	md5hash.update(line.encode())
	md5 = md5hash.hexdigest()
	logger.debug(
	"First line MD5: was %s now %s", logfile_firstlinemd5, md5
	)

	if md5 != logfile_firstlinemd5:
	# MD5 has changed, start at the beginning!
	logger.info(
	"Log file newly rolled over, starting at the beginning."
	)
	logfile_filelinepos = 0
	logfile_firstlinemd5 = md5
	else:
	logger.info(
	"Log file continues, seeking to position %d.",
	logfile_filelinepos,
	)

	if linenum < logfile_filelinepos:
	# We've seen this, carry on!
	continue

	match = LOG_FORMAT_RE.match(line)
	if not match:
	logger.debug("Not matching: %r", line)
	continue

	try:
	(jail, ip_str, ts_str) = match.groups()
	if jail in IGNORE_JAILS:
	continue

	ts = datetime.datetime.strptime(ts_str, LOG_TIMESTAMP_FORMAT)

	ips[ipaddress.ip_address(ip_str)] = int(ts.timestamp())
	except ValueError:
	# Unparseable data, continue
	logger.debug("Not parsed: %r", line, exc_info=1)
	continue

	if logfile_firstlinemd5 is None:
	# File is empty
	sys.exit(0)

	# Record our position
	logfile_lastfilepos = linenum

	# Make a note of the IPs we just saw
	NOW = int(time.time())
	for (ip, ts) in ips.items():
	# Did we have a count for this already?
	count = 0
	for (count,) in db.execute(
	"SELECT count FROM found_ips WHERE ip=?;", (str(ip),)
	):
	# Should be just one record
	break

	# New values
	count += 1
	decrement = ts + DECREMENT_DELAY

	db.execute(
	"""
	INSERT INTO found_ips
	(ip, count, lastseen, decrement)
	VALUES
	(?, ?, ?, ?)
	ON CONFLICT (ip)
	DO UPDATE SET
	count=?, lastseen=?, decrement=?;
	""",
	(str(ip), count, ts, decrement, count, ts, decrement),
	)
	logger.info(
	"%s seen %d time(s), last at %s, decrement at %s",
	ip,
	count,
	ts,
	decrement,
	)

	# Update the last-seen record
	db.execute(
	"""
	INSERT INTO lastrun
	(path, firstlinemd5, lastfilepos, mtime)
	VALUES
	(?, ?, ?, ?)
	ON CONFLICT (path)
	DO UPDATE SET
	firstlinemd5=?, lastfilepos=?, mtime=?
	""",
	(
	FAIL2BAN_LOG,
	logfile_firstlinemd5,
	logfile_lastfilepos,
	logfile_mtime_now,
	logfile_firstlinemd5,
	logfile_lastfilepos,
	logfile_mtime_now,
	),
	)
	logger.info(
	"recording file %r firstline=%s, pos=%d, mtime=%s",
	FAIL2BAN_LOG,
	logfile_firstlinemd5,
	logfile_lastfilepos,
	logfile_mtime_now,
	)

	db.commit()

	# Decrement those who are behaving
	db.execute("UPDATE found_ips SET count=count-1 WHERE decrement < ?;", (NOW,))

	# Purge those who have continued behaving
	db.execute("DELETE FROM found_ips WHERE count <= 0;")
	db.commit()

	# Coalesce misbehaving subnets
	subnets = {}
	for (ip_str, count, lastseen) in db.execute(
	"""
	SELECT ip, count, lastseen FROM found_ips;
	"""
	):
	ip = ipaddress.ip_address(ip_str)

	# IPv4 or IPv6?
	if ip.version == 4:
	min_length = MIN_SUBNET4_LENGTH
	max_length = MAX_SUBNET4_LENGTH
	else:
	min_length = MIN_SUBNET6_LENGTH
	max_length = MAX_SUBNET6_LENGTH

	for length in range(
	min_length, max_length + SUBNET_LENGTH_STEP, SUBNET_LENGTH_STEP
	):
	# strict=False → ignore host bits
	subnet = ipaddress.ip_network((ip, length), strict=False)

	try:
	(subnet_count, subnet_lastseen) = subnets[subnet]
	except KeyError:
	subnet_count = 0
	subnet_lastseen = 0

	subnet_count += count
	subnet_lastseen = max(subnet_lastseen, lastseen)
	logger.debug(
	"%s seen %d time(s) last at %s",
	subnet,
	subnet_count,
	subnet_lastseen,
	)
	subnets[subnet] = (subnet_count, subnet_lastseen)

	# Figure out who of these are misbehaving
	subnet_expiry = NOW - SUBNET_BAN_EXPIRY
	misbehaving = set()
	for (subnet, (count, lastseen)) in subnets.items():
	# IPv4 or IPv6?
	if ip.version == 4:
	max_length = MAX_SUBNET4_LENGTH
	else:
	max_length = MAX_SUBNET6_LENGTH

	# Size factor for the subnet
	size = round((max_length - subnet.prefixlen) / SUBNET_LENGTH_STEP)
	logger.debug("%s is size %d", subnet, size)

	# Compute number of strikes
	strikes = SUBNET_STRIKES + round(pow(SUBNET_STRIKES_SIZE_FACTOR, size))
	logger.debug("will accept %d strikes, has %d strikes", strikes, count)

	if (count >= strikes) and (lastseen > subnet_expiry):
	# Still misbehaving, don't bother looking at smaller subnets
	logger.info(
	"%s is misbehaving, seen %d time(s), last at %s",
	subnet,
	count,
	lastseen,
	)
	misbehaving.add(subnet)

	# Reduce overlapping subnets to the smallest possible set
	done = False
	while not done:
	done = True
	logger.debug(
	"Cleaning up overlapping subnets: %d subnet(s)", len(misbehaving)
	)
	for subnet in list(misbehaving):
	children = [
	s
	for s in misbehaving
	if (subnet is not s) and subnet.supernet_of(s)
	]
	num_children = len(children)
	if num_children == 1:
	# Keep the child, ditch the parent
	logger.debug("Keeping %s in favour of %s", children[0], subnet)
	misbehaving.discard(subnet)
	done = False
	break
	elif num_children > 1:
	# Prefer the parent since it's clear _multiple_ addresses in this
	# subnet are behaving badly.
	logger.debug(
	"Keeping %s in favour of %s",
	subnet,
	", ".join([str(s) for s in children]),
	)
	for s in children:
	misbehaving.discard(s)
	done = False
	break


	if misbehaving:
	logger.info(
	"All misbehaving subnets: %s",
	", ".join([str(s) for s in misbehaving]),
	)

	for subnet in misbehaving:
	syslog.syslog(
	syslog.LOG_NOTICE \| syslog.LOG_DAEMON,
	"%s is currently misbehaving" % subnet,
	)