README.md

qBittorrent Forwarded Port Tool

This script is used to update the qBittorrent listening port based on the current forwarded port pulled from the VPN container.
It has been tested with https://github.com/linuxserver/docker-qbittorrent and https://github.com/qdm12/gluetun. If your VPN container provides a different API for accessing the forwarded port, the class VpnControlServerApi would need to be updated to handle that API instead.

Configuration

1. Update _TORRENT_HOST and _TORRENT_PORT to match the host and port of the qBittorrent web user interface
2. Update _VPN_HOST and _VPN_CTRL_PORT to match the host and API port of the VPN container (see the wiki for API docs)
3. Change the username and password for the qBittorrent web ui to match your setup
4. Finally, make sure the Python packages qbittorrentapi and requests are installed

Usage

You can run this script manually or on a timer (such as cron or systemd). In either mode, the script gathers the current qbittorrent listening port and the current VPN forwarded port. If these do not match, the qbittorrent port is updated to match the VPN port.

port-tool.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Small utility for managing the connection port of qbittorrent in a container, setting the port to the forwarded
port of a VPN container, if they do not match
"""
import enum
import logging
import sys
from typing import Dict
from typing import Final

import qbittorrentapi
import requests


@enum.unique
class ToolExitCodes(enum.IntEnum):
    ALL_GOOD = 0
    BASE_ERROR = 1
    QBIT_AUTH_FAILURE = 2
    HTTP_ERROR = 3
    INVALID_PORT = 4
    QBIT_PREF_MISSING = 5


class VpnServerException(BaseException):
    CODE = ToolExitCodes.BASE_ERROR


class VpnServerHttpCodeException(VpnServerException):
    CODE = ToolExitCodes.HTTP_ERROR


class VpnServerInvalidPortException(VpnServerException):
    CODE = ToolExitCodes.INVALID_PORT


class VpnControlServerApi(object):
    def __init__(self,
                 host: str,
                 port: int):
        self._log = logging.getLogger(__name__)
        self._host: Final[str] = host
        self._port: Final[int] = port

        self._session = requests.Session()
        self._session.headers.update({
            "Content-Type": "application/json"
        })

        self._API_BASE: Final[str] = f"http://{self._host}:{self._port}/v1"

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self._session.close()

    def _query(self, endpoint) -> Dict:
        uri = self._API_BASE + endpoint
        self._log.debug(f"Query to {uri}")
        r = self._session.get(uri)
        if r.status_code == 200:
            self._log.debug("API query completed")
            return r.json()
        else:
            self._log.error(f"API returned {r.status_code} for {endpoint} endpoint")
            raise VpnServerHttpCodeException()

    @property
    def forwarded_port(self) -> int:
        endpoint = "/openvpn/portforwarded"
        data = self._query(endpoint)
        if "port" in data:
            vpn_forwarded_port: int = int(data["port"])
            self._log.info(f"VPN Port is {vpn_forwarded_port}")
            if 1023 < vpn_forwarded_port < 65535:
                return vpn_forwarded_port
            else:
                self._log.info(f"VPN Port invalid: {vpn_forwarded_port}")
                raise VpnServerInvalidPortException()
        else:
            self._log.info("Missing port data")
            raise VpnServerInvalidPortException()


if __name__ == "__main__":
    # Torrent host and WebUI port
    _TORRENT_HOST: Final[str] = "localhost"
    _TORRENT_PORT: Final[int] = 8282

    # VPN host and control server port
    _VPN_HOST: Final[str] = "localhost"
    _VPN_CTRL_PORT: Final[int] = 8888

    __EXIT_CODE = ToolExitCodes.ALL_GOOD

    logging.basicConfig(level=logging.INFO,
                        datefmt="%Y-%m-%d %H:%M:%S",
                        format='%(asctime)s %(name)-10s %(levelname)-8s %(message)s')

    logger = logging.getLogger("port-tool")

    qbit_port: int = -1
    vpn_port: int = -1

    # Gather the qBittorent _port
    try:

        qbt_client = qbittorrentapi.Client(host=f'http://{_TORRENT_HOST}:{_TORRENT_PORT}',
                                           username='changeme',
                                           password='whateveryouset')

        qbt_client.auth_log_in()

        logger.info(f'qBittorrent: {qbt_client.app.version}')
        logger.info(f'qBittorrent Web API: {qbt_client.app.web_api_version}')

        if "listen_port" in qbt_client.app.preferences:
            qbit_port: int = int(qbt_client.app.preferences["listen_port"])
            logger.info(f"Torrent Port is {qbit_port}")
        else:
            logger.error("Preference listen_port not found")
            __EXIT_CODE = ToolExitCodes.QBIT_PREF_MISSING

        # Gather the VPN _port
        if __EXIT_CODE == ToolExitCodes.ALL_GOOD:
            try:
                api = VpnControlServerApi(_VPN_HOST, _VPN_CTRL_PORT)
                vpn_port = api.forwarded_port
                logger.info(f"VPN port: {vpn_port}")
            except VpnServerException as e:
                logger.error(str(e))
                __EXIT_CODE = e.CODE

        # Change prefs if needed
        if __EXIT_CODE == ToolExitCodes.ALL_GOOD:
            if vpn_port != qbit_port:
                qbt_client.app.preferences = dict(listen_port=vpn_port)
                logger.info(f"Updated qBittorrent port to {vpn_port}")
            else:
                logger.info(f"Ports matched, no change ({vpn_port} == {qbit_port})")

    except qbittorrentapi.LoginFailed as e:
        logger.error(str(e))
        __EXIT_CODE = ToolExitCodes.QBIT_AUTH_FAILURE

    sys.exit(__EXIT_CODE)

@stumpylog Thanks a lot for the great script and clear explanation! 👍
I was able to test it in local setup, worked like a charm!

I'm going to adapt it further.
My local setup is on Synology 920+ NAS. I am using linuxserver/qbittorrent:4.*.*-libtorrentv1 and qmcgaw/gluetun.

I'm using also qbittorrent WebUI via https with Synology Let's encrypt certificate. With that I got en error output from script:
qbittorrentapi.exceptions.APIConnectionError: Failed to connect to qBittorrent. This is likely due to using an untrusted certificate (likely self-signed) for HTTPS qBittorrent WebUI. To suppress this error (and skip certificate verification consequently exposing the HTTPS connection to man-in-the-middle attacks), set VERIFY_WEBUI_CERTIFICATE=False when instantiating Client or set environment variable PYTHON_QBITTORRENTAPI_DO_NOT_VERIFY_WEBUI_CERTIFICATE to a non-null value. SSL Error: SSLError(MaxRetryError("HTTPSConnectionPool(host='my.host.com', port=8181): Max retries exceeded with url: /api/v2/auth/login (Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:997)')))"))

So, I need to switch to normal real SSL certificate or just use VERIFY_WEBUI_CERTIFICATE=False
Actually, my NAS isn't exposed to the internet, so I've chosen the latter :)

This part do the trick.

qbt_client = qbittorrentapi.Client(host=f'http://{_TORRENT_HOST}:{_TORRENT_PORT}',
                                           username='changeme',
                                           password='whateveryouset',
                                           VERIFY_WEBUI_CERTIFICATE=False)

For now, I've decided to pack your script into the whatever ultra-thin python image and maybe just do crontab job on NAS to run the container every 60 days or so.