"""Minimal strict bencode reader/writer for the single-file v1 teaching examples.""" import hashlib def encode(value): if isinstance(value, int): return b"i" + str(value).encode() + b"e" if isinstance(value, bytes): return str(len(value)).encode() + b":" + value if isinstance(value, list): return b"l" + b"".join(encode(v) for v in value) + b"e" if isinstance(value, dict): return b"d" + b"".join(encode(k) + encode(value[k]) for k in sorted(value)) + b"e" raise TypeError(type(value).__name__) def decode(data): """Return decoded value and the EXACT byte slice of the root dictionary's info.""" position, raw_info = 0, None def read(depth=0): nonlocal position, raw_info if depth > 100 or position >= len(data): raise ValueError("Truncated or deeply nested bencode") token = data[position:position + 1] if token == b"i": end = data.index(b"e", position) number = data[position + 1:end] value = int(number) if str(value).encode() != number: raise ValueError("Non-canonical integer") position = end + 1 return value if token in (b"l", b"d"): position += 1 result = [] if token == b"l" else {} previous = None while data[position:position + 1] != b"e": if token == b"l": result.append(read(depth + 1)) else: key = read(depth + 1) if not isinstance(key, bytes) or (previous is not None and key <= previous): raise ValueError("Dictionary keys must be unique and sorted bytes") previous = key start = position result[key] = read(depth + 1) if depth == 0 and key == b"info": raw_info = data[start:position] position += 1 return result colon = data.index(b":", position) length_bytes = data[position:colon] length = int(length_bytes) if length < 0 or str(length).encode() != length_bytes: raise ValueError("Invalid string length") position = colon + 1 value = data[position:position + length] if len(value) != length: raise ValueError("Truncated byte string") position += length return value result = read() if position != len(data): raise ValueError("Trailing data") return result, raw_info def sha1(data): return hashlib.sha1(data, usedforsecurity=False).digest() def make_metainfo(name, data, announce=b"http://127.0.0.1:6969/announce", private=False): piece_size = 65536 info = {b"name": name.encode(), b"length": len(data), b"piece length": piece_size, b"pieces": b"".join(sha1(data[i:i + piece_size]) for i in range(0, len(data), piece_size))} if private: info[b"private"] = 1 return {b"announce": announce, b"comment": b"Seminar 7: local teaching data", b"info": info} def inspect(raw): meta, raw_info = decode(raw) if not isinstance(meta, dict) or not isinstance(meta.get(b"info"), dict) or raw_info is None: raise ValueError("Missing info dictionary") info = meta[b"info"] if b"files" in info: raise ValueError("This teaching script supports single-file v1 torrents only") size, length, hashes = info[b"piece length"], info[b"length"], info[b"pieces"] if not isinstance(size, int) or size <= 0 or not isinstance(length, int) or length < 0: raise ValueError("Invalid sizes") if not isinstance(hashes, bytes) or len(hashes) != 20 * ((length + size - 1) // size): raise ValueError("Invalid piece hashes") return meta, sha1(raw_info).hex() def verify(raw, data): meta, _ = inspect(raw) info = meta[b"info"] if len(data) != info[b"length"]: return False, ["file length"] size, hashes = info[b"piece length"], info[b"pieces"] bad = [i // size for i in range(0, len(data), size) if sha1(data[i:i + size]) != hashes[(i // size) * 20:(i // size + 1) * 20]] return not bad, bad