bannedip.py

#!/usr/bin/python3
"""detect fail2ban's currently banned IP using its log file,
and show them in stdout

"""

import re
import argparse
from collections import defaultdict


REG_BAN = re.compile(r"....................... fail2ban\.actions        \[.....\]: NOTICE  \[([^\]]+)\] Ban ([0-9\.]+)")
REG_UNBAN = re.compile(r"....................... fail2ban\.actions        \[.....\]: NOTICE  \[([^\]]+)\] Unban ([0-9\.]+)")


def parse_cli() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--logfile', '-l', type=str, help='path to fail2ban log file', default='/var/log/fail2ban.log')
    return parser.parse_args()


def read_logs(lines:[str]) -> {str: str}:
    """Return map jail -> IPs found in logs,
    where IPs is a set of IP not already unbanned in associated jail"""
    banned = defaultdict(set)
    for line in map(str.strip, lines):
        if 'fail2ban.actions' in line:
            match = REG_BAN.fullmatch(line)
            if match:
                jail, ip = match.groups()
                banned[jail].add(ip)
                continue
            match = REG_UNBAN.fullmatch(line)
            if match:
                jail, ip = match.groups()
                if ip in banned[jail]:
                    banned[jail].remove(ip)
                else:
                    print(f'WARNING ip {ip} is unbanned, but not banned in the log file.')
                continue
    return {jail: ips for jail, ips in banned.items() if ips}


def test_parser():
    data = """
    2020-06-13 19:16:45,862 fail2ban.jail           [13934]: INFO    Jail 'sshd' started
    2020-06-13 19:16:45,865 fail2ban.jail           [13934]: INFO    Jail 'newjail' started
    2020-06-13 19:17:08,491 fail2ban.filter         [13934]: INFO    [newjail] Found 1.2.3.4 - 2020-06-13 19:17:07
    2020-06-13 19:17:13,525 fail2ban.filter         [13934]: INFO    [newjail] Found 1.2.3.4 - 2020-06-13 19:17:13
    2020-06-13 19:17:20,989 fail2ban.filter         [13934]: INFO    [newjail] Found 1.2.3.4 - 2020-06-13 19:17:20
    2020-06-13 19:17:21,113 fail2ban.actions        [13934]: NOTICE  [newjail] Ban 1.2.3.4
    2020-06-13 19:17:21,113 fail2ban.actions        [13934]: NOTICE  [newjail] Ban 1.2.3.5
    2020-06-13 19:18:28,472 fail2ban.actions        [13934]: NOTICE  [newjail] Unban 1.2.3.4
    2020-06-13 19:22:57,060 fail2ban.filter         [13934]: INFO    [sshd] Found 5.6.7.8 - 2020-06-13 19:22:56
    2020-06-13 19:23:07,825 fail2ban.filter         [13934]: INFO    [sshd] Found 5.6.7.8 - 2020-06-13 19:23:07
    2020-06-13 19:23:08,528 fail2ban.filter         [13934]: INFO    [sshd] Found 5.6.7.8 - 2020-06-13 19:23:08
    2020-06-13 19:23:13,709 fail2ban.filter         [13934]: INFO    [sshd] Found 5.6.7.8 - 2020-06-13 19:23:13
    2020-06-13 19:23:24,927 fail2ban.filter         [13934]: INFO    [sshd] Found 5.6.7.8 - 2020-06-13 19:23:24
    2020-06-13 19:23:24,958 fail2ban.actions        [13934]: NOTICE  [sshd] Ban 5.6.7.8
    2020-06-13 19:24:13,945 fail2ban.actions        [13934]: NOTICE  [sshd] Unban 5.6.7.8
    2020-06-13 19:23:24,958 fail2ban.actions        [13934]: NOTICE  [sshd] Ban 5.6.7.9
    """.strip()
    found = read_logs(data.splitlines())
    assert found == {'sshd': {'1.2.3.5'}, 'newjail': {'5.6.7.9'}}, found


if __name__ == "__main__":
    test_parser()
    args = parse_cli()
    with open(args.logfile) as fd:
        for jail, ips in read_logs(fd).items():
            print(jail + ':')
            for ip in ips:
                print('    ' + ip)