"""Strict BitTorrent metainfo decoding with exact info-dictionary hashing.""" from __future__ import annotations import hashlib from dataclasses import dataclass from typing import Any class BencodeError(ValueError): pass @dataclass(frozen=True) class MetaFile: path: str logical_bytes: int padding: bool @dataclass(frozen=True) class Metainfo: info_hash_v1_hex: str info_hash_v2_hex: str files: tuple[MetaFile, ...] class _Parser: def __init__(self, data: bytes): self.data = data self.position = 0 self.info_bytes: bytes | None = None def parse(self, depth: int = 0) -> Any: if depth > 100 or self.position >= len(self.data): raise BencodeError("invalid or excessively nested bencode") marker = self.data[self.position] if marker == ord("i"): return self._integer() if marker == ord("l"): return self._list(depth) if marker == ord("d"): return self._dictionary(depth) if ord("0") <= marker <= ord("9"): return self._bytes() raise BencodeError("invalid bencode marker") def _integer(self) -> int: end = self.data.find(b"e", self.position + 1) if end < 0: raise BencodeError("unterminated integer") raw = self.data[self.position + 1:end] if ( not raw or raw == b"-0" or raw.startswith(b"+") or (raw.startswith(b"0") and len(raw) > 1) or (raw.startswith(b"-0")) ): raise BencodeError("non-canonical integer") try: value = int(raw) except ValueError as exc: raise BencodeError("invalid integer") from exc self.position = end + 1 return value def _bytes(self) -> bytes: separator = self.data.find(b":", self.position) if separator < 0: raise BencodeError("invalid byte string") raw_length = self.data[self.position:separator] if not raw_length or (raw_length.startswith(b"0") and len(raw_length) > 1): raise BencodeError("non-canonical byte-string length") try: length = int(raw_length) except ValueError as exc: raise BencodeError("invalid byte-string length") from exc start = separator + 1 end = start + length if end > len(self.data): raise BencodeError("truncated byte string") self.position = end return self.data[start:end] def _list(self, depth: int) -> list[Any]: self.position += 1 result = [] while self.position < len(self.data) and self.data[self.position] != ord("e"): result.append(self.parse(depth + 1)) if self.position >= len(self.data): raise BencodeError("unterminated list") self.position += 1 return result def _dictionary(self, depth: int) -> dict[bytes, Any]: self.position += 1 result: dict[bytes, Any] = {} previous: bytes | None = None while self.position < len(self.data) and self.data[self.position] != ord("e"): key = self._bytes() if previous is not None and key <= previous: raise BencodeError("dictionary keys are duplicate or unsorted") previous = key value_start = self.position value = self.parse(depth + 1) if depth == 0 and key == b"info": self.info_bytes = self.data[value_start:self.position] result[key] = value if self.position >= len(self.data): raise BencodeError("unterminated dictionary") self.position += 1 return result def decode_metainfo(data: bytes) -> Metainfo: parser = _Parser(data) root = parser.parse() if parser.position != len(data) or not isinstance(root, dict): raise BencodeError("metainfo must be one top-level dictionary") info = root.get(b"info") if not isinstance(info, dict) or parser.info_bytes is None: raise BencodeError("metainfo has no info dictionary") is_v1 = isinstance(info.get(b"pieces"), bytes) is_v2 = info.get(b"meta version") == 2 if not is_v1 and not is_v2: raise BencodeError("unsupported torrent metainfo version") files = _v1_files(info) if is_v1 else _v2_files(info) return Metainfo( hashlib.sha1(parser.info_bytes).hexdigest() if is_v1 else "", hashlib.sha256(parser.info_bytes).hexdigest() if is_v2 else "", tuple(files), ) def encode(value: Any) -> bytes: if isinstance(value, bytes): return str(len(value)).encode("ascii") + b":" + value if isinstance(value, str): return encode(value.encode("utf-8")) if isinstance(value, bool) or not isinstance(value, (int, list, dict)): raise BencodeError("unsupported bencode value") if isinstance(value, int): return f"i{value}e".encode("ascii") if isinstance(value, list): return b"l" + b"".join(encode(item) for item in value) + b"e" keys = sorted(value) if any(not isinstance(key, bytes) for key in keys): raise BencodeError("dictionary keys must be bytes") return b"d" + b"".join(encode(key) + encode(value[key]) for key in keys) + b"e" def _v1_files(info: dict[bytes, Any]) -> list[MetaFile]: name = _component(info.get(b"name.utf-8", info.get(b"name"))) raw_files = info.get(b"files") if raw_files is None: length = _length(info.get(b"length")) return [MetaFile(name, length, _padding(info))] if not isinstance(raw_files, list): raise BencodeError("v1 files must be a list") result = [] for item in raw_files: if not isinstance(item, dict) or not isinstance(item.get(b"path"), list): raise BencodeError("invalid v1 file record") raw_path = item.get(b"path.utf-8", item[b"path"]) if not isinstance(raw_path, list): raise BencodeError("invalid v1 UTF-8 file path") components = [name] + [_component(part) for part in raw_path] result.append(MetaFile( "/".join(components), _length(item.get(b"length")), _padding(item) or _bitcomet_padding_name(components[-1]), )) return result def _v2_files(info: dict[bytes, Any]) -> list[MetaFile]: _component(info.get(b"name.utf-8", info.get(b"name"))) tree = info.get(b"file tree") if not isinstance(tree, dict): raise BencodeError("v2 metainfo has no file tree") result: list[MetaFile] = [] def visit(node: dict[bytes, Any], components: list[str]) -> None: leaf = node.get(b"") if leaf is not None: if not isinstance(leaf, dict): raise BencodeError("invalid v2 file leaf") result.append(MetaFile( "/".join(components), _length(leaf.get(b"length")), _padding(leaf), )) for key in sorted(item for item in node if item != b""): child = node[key] if not isinstance(child, dict): raise BencodeError("invalid v2 file tree node") visit(child, components + [_component(key)]) visit(tree, []) if not result: raise BencodeError("v2 file tree is empty") return result def _component(value: Any) -> str: if not isinstance(value, bytes): raise BencodeError("torrent path component must be bytes") try: decoded = value.decode("utf-8") except UnicodeDecodeError as exc: raise BencodeError("torrent path component is not UTF-8") from exc if not decoded or decoded in {".", ".."} or "/" in decoded or "\x00" in decoded: raise BencodeError("unsafe torrent path component") return decoded def _length(value: Any) -> int: if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise BencodeError("invalid torrent file length") return value def _padding(value: dict[bytes, Any]) -> bool: attributes = value.get(b"attr", b"") return isinstance(attributes, bytes) and b"p" in attributes def _bitcomet_padding_name(component: str) -> bool: """Recognize BitComet's legacy padding-file convention. Such v1 torrents often omit the standard ``attr=p`` flag, but qBittorrent/libtorrent still hides these synthetic entries from its file API. The exact reserved prefix is the interoperable marker. """ return component.startswith("_____padding_file_")