qb-ban-thunder.py

#!/usr/bin/python3

import requests, json, time

class Worker:

    API_PREFIX = 'http://10.0.2.1/qbittorrent'
    HTTP_AUTH_ENABLED = True
    HTTP_AUTH_USERNAME = 'yjsnpi'
    HTTP_AUTH_PASSWORD = '1145141919810893'
    INTERVAL_SECONDS = 5
    DEFAULT_BAN_SECONDS = 3600
    __banned_ips = {}

    def get_missions(self):
        r = requests.get('{}/api/v2/torrents/info?filter=active'.format(self.API_PREFIX), auth=self.get_auth_confidential())
        return r.json()

    def get_peers_by_mission_hash(self, mission_hash):
        r = requests.get(
            '{}/api/v2/sync/torrentPeers?hash={}'.format(self.API_PREFIX, mission_hash),
            auth=self.get_auth_confidential()
        )
        return r.json()
    
    def sumbit_banned_ips(self):
        ips = ''
        now = time.time()
        tmp_banned_ips = self.__banned_ips.copy()
        for key, value in tmp_banned_ips.items():
            if now > value['expired']:
                print('Unbanned: {} (Ban time expired)'.format(key))
                del self.__banned_ips[key]
                continue
            ips += key + '\n'
        requests.post(
            '{}/api/v2/app/setPreferences'.format(self.API_PREFIX),
            auth=self.get_auth_confidential(),
            data={
                'json': json.dumps({
                    'banned_IPs': ips
                })
            }
        )
    
    def get_auth_confidential(self):
        if self.HTTP_AUTH_ENABLED:
            return (self.HTTP_AUTH_USERNAME, self.HTTP_AUTH_PASSWORD)
        else:
            return None

    def do_once_banip(self):
        missions = self.get_missions()
        now = time.time()
        have_ip_banned = False
        for mission in missions:
            peers = self.get_peers_by_mission_hash(mission['hash'])
            for ip_port, info in peers['peers'].items():
                port_path = ip_port.rfind(':')
                ip = ''
                if ip_port.startswith('::ffff:'):
                    ip = ip_port[7:port_path]
                else:
                    ip = ip_port[:port_path]
                if info['client'].startswith('Xunlei') or info['client'].startswith('-XL'):
                    if info['progress'] == 0 and info['uploaded'] > 1000000:
                        have_ip_banned = True
                        print('Banned: {}'.format(ip))
                        self.__banned_ips[ip] = {
                            'ban_time': now,
                            'expired': now + self.DEFAULT_BAN_SECONDS
                        }
        if have_ip_banned:
            self.sumbit_banned_ips()

    def do_loop_banip(self):
        while True:
            self.do_once_banip()
            time.sleep(self.INTERVAL_SECONDS)


if __name__ == '__main__':
    worker = Worker()
    worker.do_loop_banip()