"""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_synced_peers_are_counted(self): """The live sync worker keeps its own tally like every other worker, and ?mode=syncs is the sum of them.""" before = int(self.tracker.stats("syncs").split()[0]) hashes = [self.hash_for("sync-count-%d" % i) for i in range(3)] self.tracker.livesync_send(packet(SYNC_PEER4, [(h, peer4("10.1.2.%d" % (20 + i), 6881, FLAG_SEEDING)) for i, h in enumerate(hashes)])) self.assertTrue(self.tracker.wait_until( lambda: int(self.tracker.stats("syncs").split()[0]) == before + 3), "syncs says %s" % self.tracker.stats("syncs").split()[0]) 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))