1 files changed, 308 insertions, 0 deletions

diff --git a/tests/integration/harness.py b/tests/integration/harness.py
new file mode 100644
index 0000000..03a0561
--- /dev/null
+++ b/tests/integration/harness.py
@@ -0,0 +1,308 @@
+"""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="", files=None):
+        self.port = free_port()
+        self.dir = tempfile.mkdtemp(prefix="ot-integration-")
+        self.log = None
+        self.process = None
+        try:
+            for name, content in (files or {}).items():
+                with open(os.path.join(self.dir, name), "wb") as handle:
+                    handle.write(content)
+            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.replace("{dir}", self.dir))
+            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 get_with_headers(self, path, headers=None, timeout=10):
+        """The body and the response headers, for the compressed answers."""
+        request = urllib.request.Request("http://127.0.0.1:%d%s" % (self.port, path), headers=headers or {})
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            return response.read(), dict(response.headers)
+
+    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:]}
+
+    def udp_scrape(self, sock, connection_id, hashes):
+        """One (seeders, completed, leechers) triple per hash, in order."""
+        sock.sendto(struct.pack(">QII", connection_id, 2, 3) + b"".join(hashes), ("127.0.0.1", self.port))
+        answer = sock.recv(2048)
+        action, _ = struct.unpack(">II", answer[:8])
+        assert action == 2, "udp scrape answered action %d" % action
+        return [struct.unpack(">III", answer[8 + 12 * i : 20 + 12 * i]) for i in range(len(hashes))]
+
+    def connect(self):
+        """A connection the caller drives and closes."""
+        return Connection(self.port)
+
+    def udp_raw(self, payload, timeout=1.0):
+        """Sends a packet exactly as given. Returns the answer, or None if the
+        tracker chose to say nothing, which is how it treats a broken one."""
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.settimeout(timeout)
+        try:
+            sock.sendto(payload, ("127.0.0.1", self.port))
+            try:
+                return sock.recv(2048)
+            except socket.timeout:
+                return None
+        finally:
+            sock.close()
+
+
+class Connection:
+    """One HTTP connection the caller keeps open, which urllib will not do.
+
+    Reads exactly one reply at a time, so a test can tell "the tracker
+    answered and kept the socket" from "the tracker answered and hung up"."""
+
+    def __init__(self, port, timeout=5):
+        self.sock = socket.create_connection(("127.0.0.1", port), timeout=timeout)
+        self.sock.settimeout(timeout)
+        self.buffer = b""
+
+    def send(self, path, connection="Keep-Alive", version="1.1"):
+        request = "GET %s HTTP/%s\r\nHost: tracker\r\n" % (path, version)
+        if connection is not None:
+            request += "Connection: %s\r\n" % connection
+        self.sock.sendall((request + "\r\n").encode())
+
+    def _fill(self):
+        chunk = self.sock.recv(65536)
+        if not chunk:
+            raise EOFError("the tracker closed the connection")
+        self.buffer += chunk
+
+    def reply(self):
+        """The one complete reply at the front of the stream, headers and body."""
+        while b"\r\n\r\n" not in self.buffer:
+            self._fill()
+        head, rest = self.buffer.split(b"\r\n\r\n", 1)
+        headers = {}
+        for line in head.split(b"\r\n")[1:]:
+            name, _, value = line.partition(b":")
+            headers[name.strip().lower()] = value.strip()
+
+        if headers.get(b"transfer-encoding") == b"chunked":
+            while not rest.endswith(b"0\r\n\r\n"):
+                self.buffer = head + b"\r\n\r\n" + rest
+                self._fill()
+                rest = self.buffer.split(b"\r\n\r\n", 1)[1]
+            body, self.buffer = rest, b""
+        else:
+            length = int(headers.get(b"content-length", b"0"))
+            while len(rest) < length:
+                self.buffer = head + b"\r\n\r\n" + rest
+                self._fill()
+                rest = self.buffer.split(b"\r\n\r\n", 1)[1]
+            body, self.buffer = rest[:length], rest[length:]
+        return head.split(b"\r\n")[0], headers, body
+
+    def is_open(self):
+        """True if the tracker is still holding the connection."""
+        try:
+            self.sock.settimeout(0.4)
+            return self.sock.recv(1) != b""
+        except socket.timeout:
+            return True
+        except OSError:
+            return False
+        finally:
+            try:
+                self.sock.settimeout(5)
+            except OSError:
+                pass
+
+    def close(self):
+        self.sock.close()
+
+
+class TrackerTestCase(unittest.TestCase):
+    """Gives every test its own tracker and fails it on any sanitizer output."""
+
+    config = ""
+    files = None
+
+    def setUp(self):
+        self.tracker = Tracker(self.config, self.files)
+
+    def tearDown(self):
+        # A connection the test closed is only reaped when the event loop gets
+        # round to it, and a cookie still held at exit reads as a leak. One
+        # more request puts us behind that work in the loop's queue.
+        try:
+            self.tracker.torrent_count()
+        except Exception:
+            pass
+        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]