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
|
"""The /stats endpoint.
Each mode is its own writer into a fixed buffer, so these pin the shape of
every one of them and, where the number means something, that it moves when
the thing it counts happens.
"""
import re
import unittest
import urllib.error
import xml.etree.ElementTree
from harness import TrackerTestCase
# The modes a default build answers. "tpbs" and "statedump" are full scrapes
# and belong to test_fullscrape; "woodpeckers" sits behind a
# feature flag this build does not set.
MRTG_MODES = ["peer", "conn", "scrp", "udp4", "tcp4", "torr", "fscr", "completed", "syncs"]
TEXT_MODES = ["herr", "top10", "top100", "renew"]
class StatsShape(TrackerTestCase):
def test_every_mode_answers_something(self):
for mode in MRTG_MODES + TEXT_MODES + ["everything"]:
self.assertTrue(self.tracker.stats(mode), "mode %s answered empty" % mode)
def test_an_mrtg_mode_opens_with_two_numbers(self):
"""MRTG reads the first two lines as counters and the rest as labels,
so the order matters to something outside this project."""
for mode in MRTG_MODES:
lines = self.tracker.stats(mode).splitlines()
self.assertGreaterEqual(len(lines), 2, "mode %s" % mode)
for line in lines[:2]:
self.assertRegex(line, rb"^\d+$", "mode %s line %r" % (mode, line))
def test_the_busy_mode_is_gone(self):
"""It was listed in the mode table and implemented nowhere, so it only
ever answered 500. Removed rather than written: picking poisoning
networks out stopped being meaningful once consumer lines became bot
hosts and v6 made the address space too large to watch."""
with self.assertRaises(urllib.error.HTTPError) as caught:
self.tracker.stats("busy")
self.assertEqual(caught.exception.code, 400, "an unknown mode, like any other")
def test_an_unknown_mode_is_refused(self):
with self.assertRaises(urllib.error.HTTPError) as caught:
self.tracker.stats("nosuchmode")
self.assertEqual(caught.exception.code, 400)
def test_everything_is_well_formed_xml(self):
root = xml.etree.ElementTree.fromstring(self.tracker.stats("everything").decode())
self.assertEqual(root.tag, "stats")
self.assertIsNotNone(root.find("torrents/count_mutex"))
self.assertIsNotNone(root.find("connections/tcp/announce"))
def test_the_version_mode_is_gone(self):
"""Which build a tracker is and what went into it is of use to whoever
attacks it and to nobody else on the far end of a socket. It is
reported by -v, to whoever already has the binary."""
with self.assertRaises(urllib.error.HTTPError) as caught:
self.tracker.stats("version")
self.assertEqual(caught.exception.code, 400, "an unknown mode, like any other")
def test_everything_does_not_name_the_build(self):
self.assertNotIn(b"opentracker 1", self.tracker.stats("everything"))
def test_everything_does_not_name_the_tracker_id(self):
"""The id a tracker recognises its own live sync packets by. Knowing
it is of no use to a client and of some use to whoever would like a
tracker to discard what it is told."""
self.assertNotIn(b"tracker_id", self.tracker.stats("everything"))
class StatsCount(TrackerTestCase):
def test_the_torrent_count_follows_the_torrents(self):
before = self.tracker.torrent_count()
self.tracker.announce(self.hash_for("statA"), port=6001, left=0)
self.assertEqual(self.tracker.torrent_count(), before + 1)
self.tracker.announce(self.hash_for("statB"), port=6002, left=0)
self.assertEqual(self.tracker.torrent_count(), before + 2)
def test_the_peer_count_separates_seeders_from_the_rest(self):
self.tracker.announce(self.hash_for("statpeers"), port=6001, left=0)
self.tracker.announce(self.hash_for("statpeers"), port=6002, left=4096)
peers, seeds = self.tracker.stats("peer").splitlines()[:2]
self.assertEqual(int(peers), 2)
self.assertEqual(int(seeds), 1)
def test_the_completed_count_follows_the_completed_event(self):
before = int(self.tracker.stats("completed").splitlines()[0])
self.tracker.announce(self.hash_for("statdone"), port=6001, left=0, event="completed")
self.assertEqual(int(self.tracker.stats("completed").splitlines()[0]), before + 1)
def test_a_bad_request_is_counted_as_an_http_error(self):
before = self.tracker.stats("herr")
try:
self.tracker.stats("nosuchmode")
except urllib.error.HTTPError:
pass
after = self.tracker.stats("herr")
self.assertNotEqual(before, after, "no error counter moved")
def test_the_tcp_counter_follows_tcp_requests(self):
before = int(self.tracker.stats("tcp4").splitlines()[0])
for _ in range(3):
self.tracker.announce(self.hash_for("stattcp"), port=6001, left=0)
self.assertGreaterEqual(int(self.tracker.stats("tcp4").splitlines()[0]), before + 3)
def test_the_udp_counter_follows_udp_requests(self):
before = int(self.tracker.stats("udp4").splitlines()[0])
sock, connection_id = self.tracker.udp_connect()
try:
self.tracker.udp_announce(sock, connection_id, self.hash_for("statudp"), left=0)
finally:
sock.close()
self.assertGreater(int(self.tracker.stats("udp4").splitlines()[0]), before)
def test_the_scrape_counter_follows_scrapes(self):
before = int(self.tracker.stats("scrp").splitlines()[0])
self.tracker.scrape(self.hash_for("statscrape"))
self.assertEqual(int(self.tracker.stats("scrp").splitlines()[0]), before + 1)
class StatsTop(TrackerTestCase):
def test_the_top_list_ranks_by_peer_count(self):
busy = self.hash_for("topbusy")
quiet = self.hash_for("topquiet")
for port in range(6001, 6006):
self.tracker.announce(busy, port=port, left=0)
self.tracker.announce(quiet, port=6100, left=0)
body = self.tracker.stats("top10")
self.assertIn(b"Top 10 torrents by peers", body)
ranked = re.findall(rb"\t(\d+)\t([0-9A-F]{40})", body.split(b"by seeds")[0])
self.assertEqual(ranked[0][1], busy.hex().upper().encode(), "the busy torrent ranks first")
self.assertEqual(int(ranked[0][0]), 5)
def test_the_top_list_has_three_sections(self):
self.tracker.announce(self.hash_for("topsections"), port=6001, left=0)
body = self.tracker.stats("top100")
for heading in (b"by peers", b"by seeds", b"by leechers"):
self.assertIn(heading, body)
if __name__ == "__main__":
unittest.main()
|
