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
|
"""A tracker serving both families from one socket.
These need WANT_IP_FROM_QUERY_STRING so a peer's address can be dictated
rather than being whatever loopback the test happened to come from.
"""
import os
import socket
import sys
import unittest
sys.path.insert(0, os.path.dirname(__file__))
from harness import TrackerTestCase
def addresses(answer, key):
blob, out = answer.get(key, b""), []
size = 6 if key == b"peers" else 18
for offset in range(0, len(blob), size):
entry = blob[offset : offset + size]
out.append(socket.inet_ntop(socket.AF_INET if size == 6 else socket.AF_INET6, entry[:-2]))
return out
class DualStack(TrackerTestCase):
def test_an_ipv4_peer_lands_in_the_v4_list(self):
info_hash = self.hash_for("v4only")
self.tracker.announce(info_hash, ip="10.1.2.3", port=5001, left=100)
answer = self.tracker.announce(info_hash, ip="10.1.2.4", port=5002, left=100)
self.assertEqual(sorted(addresses(answer, b"peers")), ["10.1.2.3", "10.1.2.4"])
self.assertEqual(addresses(answer, b"peers6"), [], "no v6 peer was announced")
def test_an_ipv6_peer_lands_in_the_v6_list(self):
info_hash = self.hash_for("v6only")
self.tracker.announce(info_hash, ip="2001:db8::1", port=5001, left=100)
answer = self.tracker.announce(info_hash, ip="2001:db8::2", port=5002, left=100)
self.assertEqual(sorted(addresses(answer, b"peers6")), ["2001:db8::1", "2001:db8::2"])
self.assertEqual(addresses(answer, b"peers"), [], "no v4 peer was announced")
def test_the_two_families_are_kept_apart_but_both_returned(self):
info_hash = self.hash_for("both")
self.tracker.announce(info_hash, ip="10.9.9.9", port=6001, left=100)
answer = self.tracker.announce(info_hash, ip="2001:db8::9", port=6002, left=100)
self.assertEqual(addresses(answer, b"peers"), ["10.9.9.9"])
self.assertEqual(addresses(answer, b"peers6"), ["2001:db8::9"])
def test_a_v4_mapped_address_is_the_same_peer_as_the_plain_one(self):
info_hash = self.hash_for("v4mapped")
self.tracker.announce(info_hash, ip="10.5.5.5", port=7001, left=100)
answer = self.tracker.announce(info_hash, ip="::ffff:10.5.5.5", port=7001, left=100)
self.assertEqual(answer[b"incomplete"], 1, "the same address written two ways is one peer")
self.assertEqual(addresses(answer, b"peers"), ["10.5.5.5"], "and it is held as v4")
def test_the_two_families_count_into_one_swarm(self):
info_hash = self.hash_for("counts")
self.tracker.announce(info_hash, ip="10.4.4.4", port=8001, left=0)
answer = self.tracker.announce(info_hash, ip="2001:db8::4", port=8002, left=100)
self.assertEqual(answer[b"complete"], 1)
self.assertEqual(answer[b"incomplete"], 1)
self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"complete"], 1)
def test_stopped_removes_only_the_family_it_came_from(self):
info_hash = self.hash_for("stopfamily")
self.tracker.announce(info_hash, ip="10.3.3.3", port=9001, left=100)
self.tracker.announce(info_hash, ip="2001:db8::3", port=9002, left=100)
self.tracker.announce(info_hash, ip="10.3.3.3", port=9001, left=100, event="stopped")
answer = self.tracker.announce(info_hash, ip="2001:db8::33", port=9003, left=100)
self.assertEqual(addresses(answer, b"peers"), [], "the v4 peer left")
self.assertIn("2001:db8::3", addresses(answer, b"peers6"), "the v6 one stayed")
def test_a_malformed_address_is_rejected(self):
import urllib.error
# The whole field has to be the address: a prefix that happens to
# parse would quietly record the peer somewhere else
for bad in ("999.1.2.3", "nonsense", "2001:db8:::1", "10.1.2.3.4", "10.1.2.3junk", "2001:db8::1%eth0"):
with self.assertRaises(urllib.error.HTTPError, msg="ip=%s should be refused" % bad) as caught:
self.tracker.announce(self.hash_for("badip"), ip=bad, port=6881, left=0)
self.assertEqual(caught.exception.code, 400)
if __name__ == "__main__":
unittest.main()
|
