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
|
"""The UDP tracker protocol, BEP 15.
The HTTP tests cover the torrent store; these cover the packet handling in
front of it, which has its own framing, its own connection id and its own
rules about what to answer and what to ignore.
"""
import socket
import struct
import unittest
from harness import TrackerTestCase
MAGIC = 0x41727101980
class UdpProtocol(TrackerTestCase):
def test_connect_echoes_the_transaction_id(self):
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(5)
try:
sock.sendto(struct.pack(">QII", MAGIC, 0, 0xDEADBEEF), ("127.0.0.1", self.tracker.port))
action, transaction, connection_id = struct.unpack(">IIQ", sock.recv(2048)[:16])
self.assertEqual(action, 0)
self.assertEqual(transaction, 0xDEADBEEF)
self.assertNotEqual(connection_id, 0)
finally:
sock.close()
def test_the_connection_id_is_stable_for_one_peer(self):
first, id_one = self.tracker.udp_connect()
second, id_two = self.tracker.udp_connect()
try:
self.assertEqual(id_one, id_two)
finally:
first.close()
second.close()
def test_a_wrong_connection_id_is_refused(self):
payload = struct.pack(">QII20s20sQQQIIIiH", 0, 1, 2, self.hash_for("udpbad"), b"p" * 20, 0, 0, 0, 0, 0, 0, -1, 6881)
answer = self.tracker.udp_raw(payload)
self.assertIsNotNone(answer, "expected an error packet")
action, transaction = struct.unpack(">II", answer[:8])
self.assertEqual(action, 3, "action 3 is the error packet")
self.assertIn(b"Connection ID missmatch", answer[8:])
def test_a_packet_shorter_than_the_header_is_ignored(self):
self.assertIsNone(self.tracker.udp_raw(b"\x00" * 15))
def test_an_unknown_action_is_ignored(self):
self.assertIsNone(self.tracker.udp_raw(struct.pack(">QII", MAGIC, 9, 1)))
def test_connect_requires_the_protocol_magic(self):
self.assertIsNone(self.tracker.udp_raw(struct.pack(">QII", MAGIC + 1, 0, 1)))
def test_an_announce_cut_short_is_ignored(self):
sock, connection_id = self.tracker.udp_connect()
try:
self.assertIsNone(self.tracker.udp_raw(struct.pack(">QII", connection_id, 1, 1) + b"\0" * 40))
finally:
sock.close()
class UdpAnnounce(TrackerTestCase):
def test_a_seeder_is_counted_as_a_seeder(self):
sock, connection_id = self.tracker.udp_connect()
try:
answer = self.tracker.udp_announce(sock, connection_id, self.hash_for("udpseed"), left=0)
self.assertEqual(answer["seeders"], 1)
self.assertEqual(answer["leechers"], 0)
self.assertGreater(answer["interval"], 0)
finally:
sock.close()
def test_a_leecher_is_counted_as_a_leecher(self):
sock, connection_id = self.tracker.udp_connect()
try:
answer = self.tracker.udp_announce(sock, connection_id, self.hash_for("udpleech"), left=1024)
self.assertEqual(answer["seeders"], 0)
self.assertEqual(answer["leechers"], 1)
finally:
sock.close()
def test_stopped_removes_the_peer(self):
info_hash = self.hash_for("udpstop")
sock, connection_id = self.tracker.udp_connect()
try:
self.tracker.udp_announce(sock, connection_id, info_hash, left=0)
self.tracker.udp_announce(sock, connection_id, info_hash, left=0, event=3)
answer = self.tracker.udp_announce(sock, connection_id, info_hash, left=1024)
self.assertEqual(answer["seeders"], 0, "the seeder announced stopped")
finally:
sock.close()
def test_an_announce_over_udp_is_visible_over_http(self):
"""One torrent store behind both bearers, not two."""
info_hash = self.hash_for("udpcross")
sock, connection_id = self.tracker.udp_connect()
try:
self.tracker.udp_announce(sock, connection_id, info_hash, left=0)
finally:
sock.close()
scrape = self.tracker.scrape(info_hash)
self.assertEqual(scrape[b"files"][info_hash][b"complete"], 1)
def test_the_peer_list_carries_the_announcing_peer(self):
info_hash = self.hash_for("udppeers")
sock, connection_id = self.tracker.udp_connect()
try:
answer = self.tracker.udp_announce(sock, connection_id, info_hash, left=0, port=6881)
self.assertEqual(len(answer["peers"]) % 6, 0, "v4 peers are six bytes each")
self.assertGreaterEqual(len(answer["peers"]), 6)
self.assertEqual(struct.unpack(">H", answer["peers"][4:6])[0], 6881)
finally:
sock.close()
class UdpScrape(TrackerTestCase):
def test_scrape_reports_the_swarm(self):
info_hash = self.hash_for("udpscrape")
sock, connection_id = self.tracker.udp_connect()
try:
self.tracker.udp_announce(sock, connection_id, info_hash, left=0)
seeders, completed, leechers = self.tracker.udp_scrape(sock, connection_id, [info_hash])[0]
self.assertEqual(seeders, 1)
self.assertEqual(leechers, 0)
self.assertEqual(completed, 0)
finally:
sock.close()
def test_an_unknown_hash_scrapes_as_empty(self):
sock, connection_id = self.tracker.udp_connect()
try:
self.assertEqual(self.tracker.udp_scrape(sock, connection_id, [self.hash_for("udpnone")])[0], (0, 0, 0))
finally:
sock.close()
def test_several_hashes_answer_in_order(self):
seeding = self.hash_for("udpmultiA")
leeching = self.hash_for("udpmultiB")
sock, connection_id = self.tracker.udp_connect()
try:
self.tracker.udp_announce(sock, connection_id, seeding, left=0)
self.tracker.udp_announce(sock, connection_id, leeching, left=1024)
answers = self.tracker.udp_scrape(sock, connection_id, [seeding, leeching])
self.assertEqual(answers[0][0], 1, "first hash is the seeded one")
self.assertEqual(answers[1][2], 1, "second hash is the leeched one")
finally:
sock.close()
class UdpWorkersConfiguredAfterTheListener(TrackerTestCase):
"""The harness writes its listen lines first, so this directive lands
below them: the position a config file should not care about."""
config = "listen.udp.workers 4\n"
def test_the_tracker_starts_and_still_answers_udp(self):
sock, connection_id = self.tracker.udp_connect()
try:
answer = self.tracker.udp_announce(sock, connection_id, self.hash_for("udpworkers"), left=0)
self.assertEqual(answer["seeders"], 1)
finally:
sock.close()
if __name__ == "__main__":
unittest.main()
|
