mrroman
2023-12-28 667007fe0aea7e6a22c7f1ac45dbae4a58eb672e
vilain.py
@@ -31,7 +31,7 @@
import time
CONFIGFILE = "/etc/vilain.conf"
VERSION = "0.7"
VERSION = "0.8.1"
vilain_table = "vilain_bruteforce"
LOGFILE = "/var/log/daemon"
@@ -95,7 +95,7 @@
        self.watch_while, self.default_maxtries, self.vilain_table, self.ignore_ips, self.sleeptime = load_config(config, config_dict)
        self.ip_seen_at = {}
        self.load_bad_ips()
        self.bad_ip_queue = asyncio.Queue(loop=self.loop)
        self.bad_ip_queue = asyncio.Queue()
        for entry in load_sections(config):
            logger.info("Start vilain for {}".format(entry))
@@ -183,6 +183,8 @@
            logger.info("{} detected, reason {}, count: {}, maxtries: {}".format(ip, reason, n_ip, maxtries))
            if n_ip >= maxtries:
                ret = subprocess.call(["pfctl", "-t", self.vilain_table, "-T", "add", ip])
                # now we can forget this ip
                self.ip_seen_at.pop(ip)
                logger.info("Blacklisting {}, reason {}, return code:{}".format(ip, reason, ret))
            #for debugging, this line allow us to see if the script run until here
            logger.debug('ban_ips end:{}'.format(self.ip_seen_at))
@@ -194,15 +196,11 @@
        logger.info('clean_ips started with sleeptime={}'.format(self.sleeptime))
        while True:
            await asyncio.sleep(self.watch_while)
            to_remove = []
            for recorded_ip, data in self.ip_seen_at.items():
                if time.time() - data['time'] >= self.watch_while:
                    ret = subprocess.call(["pfctl", "-t", self.vilain_table, "-T", "delete", recorded_ip])
                    logger.info("{} not blocked any more, return code:{}".format(recorded_ip, ret))
                    to_remove.append(recorded_ip)
            for ip in to_remove:
                self.ip_seen_at.pop(ip)
                    self.ip_seen_at.pop(recorded_ip)
            #for debugging, this line allow us to see if the script run until here
            ret = subprocess.call(["pfctl", "-t", self.vilain_table, "-T", "expire", self.watch_while])
            logger.debug('clean_ips end:{}'.format(self.ip_seen_at))