import logging from typing import Union, Tuple, Dict from . import abstract class TorrentParser(abstract.AbstractParser): mimetypes = {'application/x-bittorrent', } whitelist = {b'announce', b'announce-list', b'info'} def get_meta(self) -> Dict[str, str]: metadata = {} with open(self.filename, 'rb') as f: d = _BencodeHandler().bdecode(f.read()) if d is None: return {'Unknown meta': 'Unable to parse torrent file "%s".' % self.filename} for k, v in d.items(): if k not in self.whitelist: metadata[k.decode('utf-8')] = v return metadata def remove_all(self) -> bool: cleaned = dict() with open(self.filename, 'rb') as f: d = _BencodeHandler().bdecode(f.read()) if d is None: return False for k, v in d.items(): if k in self.whitelist: cleaned[k] = v with open(self.output_filename, 'wb') as f: f.write(_BencodeHandler().bencode(cleaned)) return True class _BencodeHandler(object): """ Since bencode isn't that hard to parse, MAT2 comes with its own parser, based on the spec https://wiki.theory.org/index.php/BitTorrentSpecification#Bencoding """ def __init__(self): self.__decode_func = { ord('d'): self.__decode_dict, ord('i'): self.__decode_int, ord('l'): self.__decode_list, } for i in range(0, 10): self.__decode_func[ord(str(i))] = self.__decode_string self.__encode_func = { bytes: self.__encode_string, dict: self.__encode_dict, int: self.__encode_int, list: self.__encode_list, } @staticmethod def __decode_int(s: bytes) -> Tuple[int, bytes]: s = s[1:] next_idx = s.index(b'e') if next_idx is None: raise ValueError # missing suffix if s.startswith(b'-0'): raise ValueError # negative zero doesn't exist elif s.startswith(b'0') and next_idx != 1: raise ValueError # no leading zero except for zero itself return int(s[:next_idx]), s[next_idx+1:] @staticmethod def __decode_string(s: bytes) -> Tuple[bytes, bytes]: sep = s.index(b':') if set is None: raise ValueError # missing suffix str_len = int(s[:sep]) if str_len < 0: raise ValueError elif s[0] == b'0' and sep != 1: raise ValueError s = s[1:] return s[sep:sep+str_len], s[sep+str_len:] def __decode_list(self, s: bytes) -> Tuple[list, bytes]: r = list() s = s[1:] # skip leading `l` while s[0] != ord('e'): v, s = self.__decode_func[s[0]](s) r.append(v) return r, s[1:] def __decode_dict(self, s: bytes) -> Tuple[dict, bytes]: r = dict() s = s[1:] # skip leading `d` while s[0] != ord(b'e'): k, s = self.__decode_string(s) r[k], s = self.__decode_func[s[0]](s) return r, s[1:] @staticmethod def __encode_int(x: bytes) -> bytes: return b'i' + bytes(str(x), 'utf-8') + b'e' @staticmethod def __encode_string(x: bytes) -> bytes: return bytes((str(len(x))), 'utf-8') + b':' + x def __encode_list(self, x: str) -> bytes: ret = b'' for i in x: ret += self.__encode_func[type(i)](i) return b'l' + ret + b'e' def __encode_dict(self, x: dict) -> bytes: ret = b'' for k, v in sorted(x.items()): ret += self.__encode_func[type(k)](k) ret += self.__encode_func[type(v)](v) return b'd' + ret + b'e' def bencode(self, s: Union[dict, list, bytes, int]) -> bytes: return self.__encode_func[type(s)](s) def bdecode(self, s: bytes) -> Union[dict, None]: try: r, l = self.__decode_func[s[0]](s) except (IndexError, KeyError, ValueError) as e: logging.debug("Not a valid bencoded string: %s" % e) return None if l != b'': logging.debug("Invalid bencoded value (data after valid prefix)") return None return r