1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
|
"""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 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 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):
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]
|
