"""Starts a real opentracker and talks to it over the wire. Everything here drives the binary rather than the code, so these tests cover what the unit suite cannot reach: the event loop, the HTTP parser and the torrent store working together. Run against a sanitizer build and the tracker reports its own memory errors, which the teardown turns into test failures. """ import os import shutil import signal import socket import struct import subprocess import sys import tempfile import time import unittest import urllib.error import urllib.parse import urllib.request TRACKER = os.environ.get("OPENTRACKER", os.path.join(os.path.dirname(__file__), "..", "..", "opentracker.debug")) def bdecode(data): """Enough bencode for what a tracker answers.""" def value(i): if data[i : i + 1] == b"d": i, out = i + 1, {} while data[i : i + 1] != b"e": i, key = value(i) i, out[key] = value(i) return i + 1, out if data[i : i + 1] == b"l": i, out = i + 1, [] while data[i : i + 1] != b"e": i, item = value(i) out.append(item) return i + 1, out if data[i : i + 1] == b"i": end = data.index(b"e", i) return end + 1, int(data[i + 1 : end]) colon = data.index(b":", i) length = int(data[i:colon]) return colon + 1 + length, data[colon + 1 : colon + 1 + length] return value(0)[1] def free_port(): probe = socket.socket() probe.bind(("127.0.0.1", 0)) port = probe.getsockname()[1] probe.close() return port class Tracker: """One tracker process, listening on loopback.""" def __init__(self, extra_config=""): self.port = free_port() self.dir = tempfile.mkdtemp(prefix="ot-integration-") self.log = None self.process = None try: self.config = os.path.join(self.dir, "opentracker.conf") with open(self.config, "w") as handle: handle.write("listen.tcp 127.0.0.1:%d\n" % self.port) handle.write("listen.udp 127.0.0.1:%d\n" % self.port) handle.write(extra_config) self.log = open(os.path.join(self.dir, "tracker.log"), "w+b") # Leak detection is left at the sanitizer's own default, which is on # where LeakSanitizer exists and off where it does not. Asking for it # by name makes the tracker refuse to start on macOS and FreeBSD. environment = dict(os.environ, ASAN_OPTIONS="abort_on_error=0") self.process = subprocess.Popen([TRACKER, "-f", self.config], stdout=self.log, stderr=self.log, env=environment) self._wait_for_port() except BaseException: self._cleanup() raise def _cleanup(self): """Close what we opened and take the scratch directory with it.""" if self.process is not None and self.process.poll() is None: self.process.kill() self.process.wait() if self.log is not None: self.log.close() self.log = None shutil.rmtree(self.dir, ignore_errors=True) def _wait_for_port(self): for _ in range(200): if self.process.poll() is not None: raise AssertionError("tracker exited at once:\n" + self.output()) try: probe = socket.create_connection(("127.0.0.1", self.port), timeout=0.2) probe.close() return except OSError: time.sleep(0.05) raise AssertionError("tracker never listened on %d:\n%s" % (self.port, self.output())) def output(self): if self.log is None: return "" self.log.flush() self.log.seek(0) return self.log.read().decode("utf-8", "replace") def stop(self): """Returns anything the tracker complained about on the way out. The interrupt rather than a kill is what makes the leak check work at all: LeakSanitizer reports from an exit handler, so a killed tracker reports nothing and the suite passes over leaks it never looked for. """ if self.process is not None and self.process.poll() is None: self.process.send_signal(signal.SIGINT) try: self.process.wait(timeout=5) except subprocess.TimeoutExpired: self.process.kill() self.process.wait() complaints = [ line for line in self.output().splitlines() if "Sanitizer" in line or "runtime error" in line or line.startswith("ERROR") ] self._cleanup() return complaints # ----------------------------------------------------------- over http def get(self, path, timeout=10): return urllib.request.urlopen("http://127.0.0.1:%d%s" % (self.port, path), timeout=timeout).read() def announce(self, info_hash, peer_id=b"-OT0001-aaaaaaaaaaaa", port=6881, left=0, **extra): query = { "info_hash": urllib.parse.quote_from_bytes(info_hash, safe=""), "peer_id": urllib.parse.quote_from_bytes(peer_id, safe=""), "port": port, "left": left, } query.update(extra) return bdecode(self.get("/announce?" + "&".join("%s=%s" % item for item in query.items()))) def scrape(self, info_hash): return bdecode(self.get("/scrape?info_hash=" + urllib.parse.quote_from_bytes(info_hash, safe=""))) def stats(self, mode): return self.get("/stats?mode=" + mode) def torrent_count(self): return int(self.stats("torr").split()[0]) # ------------------------------------------------------------ over udp def udp_connect(self): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.settimeout(5) sock.sendto(struct.pack(">QII", 0x41727101980, 0, 1), ("127.0.0.1", self.port)) answer = sock.recv(2048) action, _, connection_id = struct.unpack(">IIQ", answer[:16]) assert action == 0, "udp connect answered action %d" % action return sock, connection_id def udp_announce(self, sock, connection_id, info_hash, left=0, event=0, port=6881, numwant=-1): sock.sendto( struct.pack( ">QII20s20sQQQIIIiH", connection_id, 1, 2, info_hash, b"p" * 20, 0, left, 0, event, 0, 0, numwant, port ), ("127.0.0.1", self.port), ) answer = sock.recv(2048) action, _, interval, leechers, seeders = struct.unpack(">IIIII", answer[:20]) assert action == 1, "udp announce answered action %d" % action return {"interval": interval, "leechers": leechers, "seeders": seeders, "peers": answer[20:]} class TrackerTestCase(unittest.TestCase): """Gives every test its own tracker and fails it on any sanitizer output.""" config = "" def setUp(self): self.tracker = Tracker(self.config) def tearDown(self): complaints = self.tracker.stop() self.assertEqual(complaints, [], "the tracker reported:\n" + "\n".join(complaints)) def hash_for(self, name): """A distinct info hash per test, so tests cannot leak into each other.""" return (name.encode() + b"\0" * 20)[:20]