qbittorrent_ban_xunlei.py

# -*- encoding: utf-8 -*-
# Author: Epix
#
# Ban Xunlei IP for qBittorrent
# This script will check qBittorrent connected peers every 30 seconds.
# When an IP with UA in `BAN_PATTERNS` is found, the IP will be banned.
# However, if the IP is uploading or reporting 100% progress, it will say "WTF?"


import re
from functools import lru_cache
from time import sleep

import qbittorrentapi
from loguru import logger

QBT_INFO = {'host': '10.0.0.234', 'port': 8080}

# UA patterns to ban
BAN_PATTERNS = [
    re.compile(r'Xunlei .*'),
    re.compile(r'7\.10\.\d+\.\d+'),
    re.compile(r'7\.9\.\d+\.\d+'),
    re.compile(r'-XL0012-.*'),
]


@lru_cache(maxsize=1)
def get_client():
    return qbittorrentapi.Client(**QBT_INFO)


def check():
    qbt_client = get_client()
    # bad_ips = set()
    for torrent in qbt_client.torrents_info(status_filter='active'):
        torrent_hash = torrent.info.hash
        peers = qbt_client.sync_torrent_peers(hash=torrent_hash).peers
        for peer in peers.values():
            peer_client = peer.client
            peer_client = peer_client.encode('ascii', 'ignore').decode('ascii')
            peer_client = re.sub(r'\s+', ' ', peer_client)
            for pattern in BAN_PATTERNS:
                if pattern.match(peer_client):
                    dl_speed = peer.dl_speed
                    progress = peer['progress']
                    if dl_speed > 0 or progress >= 0.9999:
                        logger.error(f'WTF client {peer_client} is uploading at {dl_speed} Bytes/s')
                    else:
                        bad_ip_port = f'{peer.ip}:{peer.port}'
                        # bad_ips.add(bad_ip)
                        qbt_client.transfer.ban_peers(peers=bad_ip_port)
                        logger.warning(f'Found bad client {peer_client} from IP {bad_ip_port}')
                    break


def main():
    while True:
        check()
        sleep(30)


if __name__ == '__main__':
    main()