diff --git a/lbry/dht/peer.py b/lbry/dht/peer.py index 19fbb92e3..bf2c0deab 100644 --- a/lbry/dht/peer.py +++ b/lbry/dht/peer.py @@ -33,12 +33,14 @@ def is_valid_public_ipv4(address, allow_localhost: bool = False): parsed_ip = ipaddress.ip_address(address) if parsed_ip.is_loopback and allow_localhost: return True - return not any((parsed_ip.version != 4, parsed_ip.is_unspecified, parsed_ip.is_link_local, - parsed_ip.is_loopback, parsed_ip.is_multicast, parsed_ip.is_reserved, parsed_ip.is_private, - parsed_ip.is_reserved, - CARRIER_GRADE_NAT_SUBNET.supernet_of(ipaddress.ip_network(f"{address}/32")), - IPV4_TO_6_RELAY_SUBNET.supernet_of(ipaddress.ip_network(f"{address}/32")))) - except ipaddress.AddressValueError: + + if any((parsed_ip.version != 4, parsed_ip.is_unspecified, parsed_ip.is_link_local, parsed_ip.is_loopback, + parsed_ip.is_multicast, parsed_ip.is_reserved, parsed_ip.is_private, parsed_ip.is_reserved)): + return False + else: + return not any((CARRIER_GRADE_NAT_SUBNET.supernet_of(ipaddress.ip_network(f"{address}/32")), + IPV4_TO_6_RELAY_SUBNET.supernet_of(ipaddress.ip_network(f"{address}/32")))) + except (ipaddress.AddressValueError, ValueError): return False diff --git a/tests/unit/dht/test_peer.py b/tests/unit/dht/test_peer.py index 8bf5d70a9..8a61dad14 100644 --- a/tests/unit/dht/test_peer.py +++ b/tests/unit/dht/test_peer.py @@ -1,7 +1,7 @@ import asyncio import unittest from lbry.utils import generate_id -from lbry.dht.peer import PeerManager, make_kademlia_peer +from lbry.dht.peer import PeerManager, make_kademlia_peer, is_valid_public_ipv4 from lbry.testcase import AsyncioTestCase @@ -50,6 +50,31 @@ class PeerTest(AsyncioTestCase): for i in range(32): self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], f"{224 + i}.0.0.0", 1000) self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '255.255.255.255', 1000) + self.assertRaises( + ValueError, make_kademlia_peer, self.node_ids[1], 'beee:eeee:eeee:eeee:eeee:eeee:eeee:eeef', 1000 + ) + self.assertRaises( + ValueError, make_kademlia_peer, self.node_ids[1], '2001:db8::ff00:42:8329', 1000 + ) + + def test_is_valid_ipv4(self): + self.assertFalse(is_valid_public_ipv4('beee:eeee:eeee:eeee:eeee:eeee:eeee:eeef')) + self.assertFalse(is_valid_public_ipv4('beee:eeee:eeee:eeee:eeee:eeee:eeee:eeef', True)) + + self.assertFalse(is_valid_public_ipv4('2001:db8::ff00:42:8329')) + self.assertFalse(is_valid_public_ipv4('2001:db8::ff00:42:8329', True)) + + self.assertFalse(is_valid_public_ipv4('127.0.0.1')) + self.assertTrue(is_valid_public_ipv4('127.0.0.1', True)) + + self.assertFalse(is_valid_public_ipv4('172.16.0.1')) + self.assertFalse(is_valid_public_ipv4('172.16.0.1', True)) + + self.assertTrue(is_valid_public_ipv4('1.2.3.4')) + self.assertTrue(is_valid_public_ipv4('1.2.3.4', True)) + + self.assertFalse(is_valid_public_ipv4('derp')) + self.assertFalse(is_valid_public_ipv4('derp', True)) def test_boolean(self): self.assertNotEqual(self.first_contact, self.second_contact)