"""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()