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
|
"""Live sync, driven over the multicast group a cluster would use.
Live sync sits behind WANT_SYNC_LIVE, so these only run against a binary
built with it: make check-livesync. Against any other build they skip, which
is what the feature list in -v is for.
"""
import socket
import struct
import subprocess
import unittest
from harness import TRACKER, TrackerTestCase
# enum { OT_SYNC_PEER4, OT_SYNC_PEER6 } in src/ot_livesync.c
SYNC_PEER4 = 0
SYNC_PEER6 = 1
# src/trackerlogic.h
FLAG_SEEDING = 0x80
FLAG_COMPLETED = 0x40
FLAG_STOPPED = 0x20
def has_livesync():
done = subprocess.run([TRACKER, "-v"], capture_output=True, timeout=30)
return b"WANT_SYNC_LIVE" in done.stderr
def peer4(address, port, flags):
"""Four bytes of address, two of port, one of flags, one the tracker
overwrites with its own idea of the time."""
return socket.inet_aton(address) + struct.pack("!H", port) + bytes([flags, 0])
def peer6(address, port, flags):
return socket.inet_pton(socket.AF_INET6, address) + struct.pack("!H", port) + bytes([flags, 0])
def packet(kind, entries, tracker_id=0xDEADBEEF):
body = b"".join(info_hash + peer for info_hash, peer in entries)
return struct.pack("!I", tracker_id) + struct.pack("!I", kind) + body
@unittest.skipUnless(has_livesync(), "this build has no WANT_SYNC_LIVE")
class LiveSyncIn(TrackerTestCase):
config = "livesync.cluster.listen 127.0.0.1:{livesync_port}\nlivesync.cluster.node_ip 127.0.0.1\n"
def counts(self, info_hash):
"""Seeders and leechers a scrape reports for one torrent."""
files = self.tracker.scrape(info_hash)[b"files"]
if info_hash not in files:
return None
entry = files[info_hash]
return entry[b"complete"], entry[b"incomplete"]
def test_a_synced_seeder_arrives(self):
info_hash = self.hash_for("sync-seed")
self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.2.3", 6881, FLAG_SEEDING))]))
self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) == (1, 0)),
"scrape says %r" % (self.counts(info_hash),))
def test_a_synced_leecher_arrives(self):
info_hash = self.hash_for("sync-leech")
self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.2.4", 6881, 0))]))
self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) == (0, 1)),
"scrape says %r" % (self.counts(info_hash),))
def test_a_v6_peer_arrives(self):
info_hash = self.hash_for("sync-v6")
self.tracker.livesync_send(packet(SYNC_PEER6, [(info_hash, peer6("2001:db8::1", 6881, FLAG_SEEDING))]))
self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) == (1, 0)),
"scrape says %r" % (self.counts(info_hash),))
def test_one_packet_carries_several_torrents(self):
hashes = [self.hash_for("sync-many-%d" % i) for i in range(4)]
self.tracker.livesync_send(packet(SYNC_PEER4, [(h, peer4("10.1.2.%d" % (5 + i), 6881, FLAG_SEEDING))
for i, h in enumerate(hashes)]))
self.assertTrue(self.tracker.wait_until(lambda: all(self.counts(h) == (1, 0) for h in hashes)))
def test_a_stopped_peer_is_taken_away_again(self):
info_hash = self.hash_for("sync-stop")
peer = peer4("10.1.2.9", 6881, FLAG_SEEDING)
self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer)]))
self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) == (1, 0)))
self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.2.9", 6881, FLAG_STOPPED))]))
self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) in (None, (0, 0))),
"scrape still says %r" % (self.counts(info_hash),))
def test_a_truncated_entry_is_dropped(self):
"""The loop takes whole hash and peer pairs and leaves a short tail
where it is, so the complete entry before it still lands."""
good, short = self.hash_for("sync-good"), self.hash_for("sync-short")
payload = packet(SYNC_PEER4, [(good, peer4("10.1.2.10", 6881, FLAG_SEEDING))]) + short + b"\x0a\x01"
self.tracker.livesync_send(payload)
self.assertTrue(self.tracker.wait_until(lambda: self.counts(good) == (1, 0)))
self.assertIsNone(self.counts(short))
def test_an_unknown_packet_type_is_ignored(self):
info_hash = self.hash_for("sync-unknown")
self.tracker.livesync_send(packet(99, [(info_hash, peer4("10.1.2.11", 6881, FLAG_SEEDING))]))
self.assertFalse(self.tracker.wait_until(lambda: self.counts(info_hash) is not None, timeout=1))
def test_a_packet_with_nothing_but_a_header_is_ignored(self):
before = self.tracker.torrent_count()
self.tracker.livesync_send(struct.pack("!I", 0xDEADBEEF) + struct.pack("!I", SYNC_PEER4))
self.assertFalse(self.tracker.wait_until(lambda: self.tracker.torrent_count() != before, timeout=1))
@unittest.skipUnless(has_livesync(), "this build has no WANT_SYNC_LIVE")
class LiveSyncAccess(TrackerTestCase):
"""Nothing blesses 127.0.0.1 here, so the tracker should take nothing
from it. Source address checking is all live sync has."""
config = "livesync.cluster.listen 127.0.0.1:{livesync_port}\nlivesync.cluster.node_ip 10.0.0.1\n"
def test_an_unblessed_sender_is_ignored(self):
info_hash = self.hash_for("sync-unblessed")
self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.2.12", 6881, FLAG_SEEDING))]))
self.assertFalse(self.tracker.wait_until(lambda: self.tracker.torrent_count() != 0, timeout=1))
|
