[Experimental] Added direct injection for Deluge (#5)

* [WIP] getting started with Deluge connection

* Finished basic PoC for deluge injection

* Added tests for deluge client

* reverted main.py

* Updated deluge to allow specifying a save_path_override

* Hooked up injection to torrent scanner

* Doing a little renaming

* set save_path_override to parent directory of files

* Output torrents are now stored in a namespaced directory

* Lots of tests; huge refactor

* Removed optional keys from default config

.gitignore

@@ -132,6 +132,10 @@ cython_debug/
 .vscode/
 
 .env
-tests/support/torrents/example.torrent
 scratchpad.md
 .DS_Store
+tmp/
+
+*.torrent
+!tests/support/files/*.torrent
+tests/support/files/example.torrent

docker-compose.yaml

@@ -17,3 +17,5 @@ services:
         wait
     stdin_open: true
     tty: true
+    env_file:
+      - .env

main.py

@@ -4,21 +4,22 @@
 from src.args import parse_args
 from src.config import Config
 from src.scanner import scan_torrent_directory, scan_torrent_file
-
 from src.webserver import run_webserver
+from src.injection import Injection
 
 
 def cli_entrypoint(args):
   try:
     config = Config().load(args.config_file)
     red_api, ops_api = __verify_api_keys(config)
+    injector = Injection(config).setup() if config.inject_torrents else None
 
     if args.server:
-      run_webserver(args.input_directory, args.output_directory, red_api, ops_api, port=config.server_port)
+      run_webserver(args.input_directory, args.output_directory, red_api, ops_api, injector, port=config.server_port)
     elif args.input_file:
-      print(scan_torrent_file(args.input_file, args.output_directory, red_api, ops_api))
+      print(scan_torrent_file(args.input_file, args.output_directory, red_api, ops_api, injector))
     elif args.input_directory:
-      print(scan_torrent_directory(args.input_directory, args.output_directory, red_api, ops_api))
+      print(scan_torrent_directory(args.input_directory, args.output_directory, red_api, ops_api, injector))
   except Exception as e:
     print(f"{Fore.RED}{str(e)}{Fore.RESET}")
     exit(1)

src/clients/deluge.py

@@ -0,0 +1,157 @@
+import json
+import base64
+import requests
+from pathlib import Path
+
+from ..errors import TorrentClientError
+from .torrent_client import TorrentClient
+from requests.exceptions import RequestException
+from requests.structures import CaseInsensitiveDict
+
+
+class Deluge(TorrentClient):
+  def __init__(self, rpc_url):
+    super().__init__()
+    self._rpc_url = rpc_url
+    self._deluge_cookie = None
+    self._deluge_request_id = 0
+    self._label_plugin_enabled = False
+
+  def setup(self):
+    connection_response = self.__authenticate()
+    self._label_plugin_enabled = self.__is_label_plugin_enabled()
+
+    return connection_response
+
+  def get_torrent_info(self, infohash):
+    infohash = infohash.lower()
+    params = [
+      [
+        "name",
+        "state",
+        "progress",
+        "save_path",
+        "label",
+        "total_remaining",
+      ],
+      {"hash": infohash},
+    ]
+
+    response = self.__request("web.update_ui", params)
+    if "torrents" in response:
+      torrent = response["torrents"].get(infohash)
+
+      if torrent is None:
+        raise TorrentClientError(f"Torrent not found in client ({infohash})")
+    else:
+      raise TorrentClientError("Client returned unexpected response (object missing)")
+
+    torrent_completed = (
+      (torrent["state"] == "Paused" and (torrent["progress"] == 100 or not torrent["total_remaining"]))
+      or torrent["state"] == "Seeding"
+      or torrent["progress"] == 100
+      or not torrent["total_remaining"]
+    )
+
+    return {
+      "complete": torrent_completed,
+      "label": torrent.get("label"),
+      "save_path": torrent["save_path"],
+    }
+
+  def inject_torrent(self, source_torrent_infohash, new_torrent_filepath, save_path_override=None):
+    source_torrent_info = self.get_torrent_info(source_torrent_infohash)
+
+    if not source_torrent_info["complete"]:
+      raise TorrentClientError("Cannot inject a torrent that is not complete")
+
+    params = [
+      f"{Path(new_torrent_filepath).stem}.fertilizer.torrent",
+      base64.b64encode(open(new_torrent_filepath, "rb").read()).decode(),
+      {
+        "download_location": save_path_override if save_path_override else source_torrent_info["save_path"],
+        "seed_mode": True,
+        "add_paused": False,
+      },
+    ]
+
+    new_torrent_infohash = self.__request("core.add_torrent_file", params)
+    newtorrent_label = self.__determine_label(source_torrent_info)
+    self.__set_label(new_torrent_infohash, newtorrent_label)
+
+    return new_torrent_infohash
+
+  def __authenticate(self):
+    _href, _username, password = self._extract_credentials_from_url(self._rpc_url)
+    if not password:
+      raise Exception("You need to define a password in the Deluge RPC URL. (e.g. http://:<PASSWORD>@localhost:8112)")
+
+    auth_response = self.__request("auth.login", [password])
+    if not auth_response:
+      raise TorrentClientError("Reached Deluge RPC endpoint but failed to authenticate")
+
+    return self.__request("web.connected")
+
+  def __is_label_plugin_enabled(self):
+    response = self.__request("core.get_enabled_plugins")
+
+    return "Label" in response
+
+  def __determine_label(self, torrent_info):
+    current_label = torrent_info.get("label")
+
+    if not current_label or current_label == self.torrent_label:
+      return self.torrent_label
+
+    return f"{current_label}.{self.torrent_label}"
+
+  def __set_label(self, infohash, label):
+    if not self._label_plugin_enabled:
+      return
+
+    current_labels = self.__request("label.get_labels")
+    if label not in current_labels:
+      self.__request("label.add", [label])
+
+    return self.__request("label.set_torrent", [infohash, label])
+
+  def __request(self, method, params=[]):
+    href, _, _ = self._extract_credentials_from_url(self._rpc_url)
+
+    headers = CaseInsensitiveDict()
+    headers["Content-Type"] = "application/json"
+    if self._deluge_cookie:
+      headers["Cookie"] = self._deluge_cookie
+
+    try:
+      response = requests.post(
+        href,
+        json={
+          "method": method,
+          "params": params,
+          "id": self._deluge_request_id,
+        },
+        headers=headers,
+        timeout=10,
+      )
+      self._deluge_request_id += 1
+    except RequestException as network_error:
+      if network_error.response and network_error.response.status_code == 408:
+        raise TorrentClientError(f"Deluge method {method} timed out after 10 seconds")
+      raise TorrentClientError(f"Failed to connect to Deluge at {href}") from network_error
+
+    try:
+      json_response = response.json()
+    except json.JSONDecodeError as json_parse_error:
+      raise TorrentClientError(f"Deluge method {method} response was non-JSON") from json_parse_error
+
+    self.__handle_response_headers(response.headers)
+
+    if "error" in json_response and json_response["error"]:
+      raise TorrentClientError(f"Deluge method {method} returned an error: {json_response['error']}")
+
+    return json_response["result"]
+
+  def __handle_response_headers(self, headers):
+    if "Set-Cookie" in headers:
+      self._deluge_cookie = headers["Set-Cookie"].split(";")[0]

src/clients/torrent_client.py

@@ -0,0 +1,15 @@
+from urllib.parse import urlparse, unquote
+
+
+class TorrentClient:
+  def __init__(self):
+    self.torrent_label = "fertilizer"
+
+  def _extract_credentials_from_url(self, url):
+    parsed_url = urlparse(url)
+    username = unquote(parsed_url.username) if parsed_url.username else ""
+    password = unquote(parsed_url.password) if parsed_url.password else ""
+    origin = f"{parsed_url.scheme}://{parsed_url.hostname}:{parsed_url.port}"
+    href = origin + (parsed_url.path if parsed_url.path != "/" else "")
+
+    return href, username, password

src/config.py

@@ -31,13 +31,25 @@ def ops_key(self) -> str:
 
   @property
   def server_port(self) -> str:
-    return self.__get_key("port", "9713")
+    return self.__get_key("port", must_exist=False) or "9713"
 
-  def __get_key(self, key, default=None):
+  @property
+  def deluge_rpc_url(self) -> str | None:
+    return self.__get_key("deluge_rpc_url", must_exist=False) or None
+
+  @property
+  def inject_torrents(self) -> str | bool:
+    return self.__get_key("inject_torrents", must_exist=False) or False
+
+  @property
+  def injection_link_directory(self) -> str | None:
+    return self.__get_key("injection_link_directory", must_exist=False) or None
+
+  def __get_key(self, key, must_exist=True):
     try:
       return self._json[key]
     except KeyError:
-      if default is not None:
-        return default
+      if must_exist:
+        raise ConfigKeyError(f"Key '{key}' not found in config file.")
 
-      raise ConfigKeyError(f"Key '{key}' not found in config file.")
+      return None

src/errors.py

@@ -44,3 +44,11 @@ class TorrentAlreadyExistsError(Exception):
 
 class ConfigKeyError(Exception):
   pass
+
+
+class TorrentClientError(Exception):
+  pass
+
+
+class TorrentInjectionError(Exception):
+  pass

src/injection.py

@@ -0,0 +1,101 @@
+import os
+import shutil
+
+from .errors import TorrentInjectionError
+from .clients.deluge import Deluge
+from .config import Config
+from .parser import calculate_infohash, get_name, get_torrent_data
+
+
+class Injection:
+  def __init__(self, config: Config):
+    self.config = self.__validate_config(config)
+    self.linking_directory = config.injection_link_directory
+    self.client = self.__determine_torrent_client(config)
+
+  def setup(self):
+    self.client.setup()
+    return self
+
+  def inject_torrent(self, source_torrent_filepath, new_torrent_filepath, new_tracker):
+    source_torrent_data = get_torrent_data(source_torrent_filepath)
+    source_torrent_file_or_dir = self.__determine_torrent_data_location(source_torrent_data)
+    output_location = self.__determine_output_location(source_torrent_file_or_dir, new_tracker)
+    self.__link_files_to_output_location(source_torrent_file_or_dir, output_location)
+    output_parent_directory = os.path.dirname(os.path.normpath(output_location))
+
+    return self.client.inject_torrent(
+      calculate_infohash(source_torrent_data),
+      new_torrent_filepath,
+      save_path_override=output_parent_directory,
+    )
+
+  def __validate_config(self, config: Config):
+    if not config.inject_torrents:
+      raise TorrentInjectionError("Torrent injection is disabled in the config file.")
+
+    if not config.injection_link_directory:
+      raise TorrentInjectionError("No injection link directory specified in the config file.")
+
+    # NOTE: will add more checks here as more clients get added
+    if not config.deluge_rpc_url:
+      raise TorrentInjectionError("No torrent client configuration specified in the config file.")
+
+    return config
+
+  def __determine_torrent_client(self, config: Config):
+    # NOTE: will add more conditions here as more clients get added
+    if config.deluge_rpc_url:
+      return Deluge(config.deluge_rpc_url)
+
+  # If the torrent is a single bare file, this returns the path _to that file_
+  # If the torrent is one or many files in a directory, this returns the topmost directory path
+  def __determine_torrent_data_location(self, torrent_data):
+    # Note on torrent file structures:
+    # --------
+    # From my testing, all torrents have a `name` stored at `[b"info"][b"name"]`. This appears to always
+    # be the name of the top-most file or directory that contains torrent data. Although I've always seen
+    # the name key, apparently it is only a suggestion so we add checks to verify existence of the file/directory
+    # (although we do nothing to try and recover if that file/directory is missing)
+    #
+    # So if the torrent is a single file, the `name` is the full filename of that file, including extension.
+    # If the torrent contains a directory, the `name` is the name of that directory and the subfiles of the
+    # torrents are stored under that directory.
+    #
+    # If a torrent has one file and that file is at the root level of the torrent, the `files` key is absent.
+    # If a torrent has multiple files OR a single file but it's in a directory, the `files` key is present
+    # and is an array of dictionaries. Each dictionary has a `path` key that is an array of bytestrings where
+    # each array member is a part of the path to the file. In other words, if you joined all the bytestrings
+    # in the `path` array for a given file, you'd get the path to the file relative to the topmost parent
+    # directory (which in our case is the `name`).
+    #
+    # See also: https://en.wikipedia.org/wiki/Torrent_file#File_struct
+    infohash = calculate_infohash(torrent_data)
+    torrent_info_from_client = self.client.get_torrent_info(infohash)
+    client_save_path = torrent_info_from_client["save_path"]
+    torrent_name = get_name(torrent_data).decode()
+    proposed_torrent_data_location = os.path.join(client_save_path, torrent_name)
+
+    if os.path.exists(proposed_torrent_data_location):
+      return proposed_torrent_data_location
+
+    raise TorrentInjectionError(
+      f"Could not determine the location of the torrent data: {proposed_torrent_data_location}"
+    )
+
+  def __determine_output_location(self, source_torrent_file_or_dir, new_tracker):
+    tracker_output_directory = os.path.join(self.linking_directory, new_tracker)
+    os.makedirs(tracker_output_directory, exist_ok=True)
+
+    return os.path.join(tracker_output_directory, os.path.basename(source_torrent_file_or_dir))
+
+  def __link_files_to_output_location(self, source_torrent_file_or_dir, output_location):
+    if os.path.exists(output_location):
+      raise TorrentInjectionError(f"Cannot link given torrent since it's already been linked: {output_location}")
+
+    if os.path.isfile(source_torrent_file_or_dir):
+      os.link(source_torrent_file_or_dir, output_location)
+    elif os.path.isdir(source_torrent_file_or_dir):
+      shutil.copytree(source_torrent_file_or_dir, output_location, copy_function=os.link)
+
+    return output_location

src/parser.py

@@ -1,3 +1,4 @@
+import os
 import copy
 import bencoder
 from hashlib import sha1
@@ -21,6 +22,13 @@ def get_source(torrent_data: dict) -> bytes:
     return None
 
 
+def get_name(torrent_data: dict) -> bytes:
+  try:
+    return torrent_data[b"info"][b"name"]
+  except KeyError:
+    return None
+
+
 def get_announce_url(torrent_data: dict) -> bytes:
   try:
     return torrent_data[b"announce"]
@@ -61,8 +69,11 @@ def get_torrent_data(filename: str) -> dict:
     return None
 
 
-def save_torrent_data(filename: str, torrent_data: dict) -> str:
-  with open(filename, "wb") as f:
+def save_torrent_data(filepath: str, torrent_data: dict) -> str:
+  parent_dir = os.path.dirname(filepath)
+  os.makedirs(parent_dir, exist_ok=True)
+
+  with open(filepath, "wb") as f:
     f.write(bencoder.encode(torrent_data))
 
-  return filename
+  return filepath