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
|
"""The blacklist, loaded from a file at startup.
A blacklisted info hash must be refused at both the announce and the scrape
entry, and the file parser has to cope with the shapes an operator actually
produces: an empty file, and a last line with no newline after it.
"""
import struct
import unittest
from harness import TrackerTestCase
BLOCKED = b"\xb1" * 20
ALLOWED = b"\xa1" * 20
def listing(*hashes):
return b"".join(h.hex().upper().encode() + b"\n" for h in hashes)
class Blacklist(TrackerTestCase):
config = "access.blacklist {dir}/blacklist\n"
files = {"blacklist": listing(BLOCKED)}
def test_a_blacklisted_hash_is_refused(self):
answer = self.tracker.announce(BLOCKED, port=6881, left=0)
self.assertIn(b"failure reason", answer)
self.assertIn(b"not authorized", answer[b"failure reason"])
def test_a_blacklisted_hash_never_becomes_a_torrent(self):
before = self.tracker.torrent_count()
self.tracker.announce(BLOCKED, port=6881, left=0)
self.assertEqual(self.tracker.torrent_count(), before, "the refused hash was stored anyway")
def test_any_other_hash_is_served(self):
answer = self.tracker.announce(ALLOWED, port=6881, left=0)
self.assertNotIn(b"failure reason", answer)
self.assertEqual(answer[b"complete"], 1)
def test_the_block_is_an_error_packet_over_udp(self):
"""BEP 15 has an error action, which is what a refusal should use.
It used to answer the announce header with no body behind it, a
packet too short for any client to read."""
sock, connection_id = self.tracker.udp_connect()
sock.close()
request = struct.pack(
">QII20s20sQQQIIIiH", connection_id, 1, 7, BLOCKED, b"p" * 20, 0, 0, 0, 0, 0, 0, -1, 6881
)
answer = self.tracker.udp_raw(request)
self.assertIsNotNone(answer)
action, transaction = struct.unpack(">II", answer[:8])
self.assertEqual(action, 3, "action 3 is the error packet")
self.assertEqual(transaction, 7)
self.assertIn(b"not authorized", answer[8:])
def test_a_blocked_hash_is_not_stored_when_it_arrives_over_udp(self):
sock, connection_id = self.tracker.udp_connect()
sock.close()
self.tracker.udp_raw(
struct.pack(">QII20s20sQQQIIIiH", connection_id, 1, 8, BLOCKED, b"p" * 20, 0, 0, 0, 0, 0, 0, -1, 6881)
)
self.assertEqual(self.tracker.torrent_count(), 0)
class BlacklistWithoutTrailingNewline(TrackerTestCase):
"""A single hash with no newline after it used to size the list at zero
and then write into it, which took the tracker down before it listened."""
config = "access.blacklist {dir}/blacklist\n"
files = {"blacklist": BLOCKED.hex().upper().encode()}
def test_the_last_line_counts_even_without_its_newline(self):
answer = self.tracker.announce(BLOCKED, port=6881, left=0)
self.assertIn(b"failure reason", answer)
def test_and_the_tracker_still_serves_everything_else(self):
self.assertEqual(self.tracker.announce(ALLOWED, port=6881, left=0)[b"complete"], 1)
class EmptyBlacklist(TrackerTestCase):
config = "access.blacklist {dir}/blacklist\n"
files = {"blacklist": b""}
def test_an_empty_file_blocks_nothing_and_starts_cleanly(self):
self.assertEqual(self.tracker.announce(BLOCKED, port=6881, left=0)[b"complete"], 1)
self.assertEqual(self.tracker.announce(ALLOWED, port=6882, left=0)[b"complete"], 1)
class MissingBlacklist(TrackerTestCase):
"""The file is named but absent: the tracker warns and serves everything,
because it may be created later."""
config = "access.blacklist {dir}/absent\n"
def test_a_missing_file_blocks_nothing(self):
self.assertEqual(self.tracker.announce(BLOCKED, port=6881, left=0)[b"complete"], 1)
if __name__ == "__main__":
unittest.main()
|
