Merge pull request #2649 from lbryio/invalid-ips

Fix handling reserved ips

lbry/lbry/dht/peer.py

@@ -15,14 +15,29 @@ log = logging.getLogger(__name__)
 @lru_cache(1024)
 def make_kademlia_peer(node_id: typing.Optional[bytes], address: typing.Optional[str],
                        udp_port: typing.Optional[int] = None,
-                       tcp_port: typing.Optional[int] = None) -> 'KademliaPeer':
+                       tcp_port: typing.Optional[int] = None,
-    return KademliaPeer(address, node_id, udp_port, tcp_port=tcp_port)
+                       allow_localhost: bool = False) -> 'KademliaPeer':
+    return KademliaPeer(address, node_id, udp_port, tcp_port=tcp_port, allow_localhost=allow_localhost)
 
 
-def is_valid_ipv4(address):
+# the ipaddress module does not show these subnets as reserved
+carrier_grade_NAT_subnet = ipaddress.ip_network('100.64.0.0/10')
+ip4_to_6_relay_subnet = ipaddress.ip_network('192.88.99.0/24')
+
+ALLOW_LOCALHOST = False
+
+
+def is_valid_public_ipv4(address, allow_localhost: bool = False):
+    allow_localhost = bool(allow_localhost or ALLOW_LOCALHOST)
     try:
-        ip = ipaddress.ip_address(address)
+        parsed_ip = ipaddress.ip_address(address)
-        return ip.version == 4
+        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")),
+                        ip4_to_6_relay_subnet.supernet_of(ipaddress.ip_network(f"{address}/32"))))
     except ipaddress.AddressValueError:
         return False
 
@@ -151,6 +166,7 @@ class KademliaPeer:
     udp_port: typing.Optional[int] = field(hash=True)
     tcp_port: typing.Optional[int] = field(compare=False, hash=False)
     protocol_version: typing.Optional[int] = field(default=1, compare=False, hash=False)
+    allow_localhost: bool = field(default=False, compare=False, hash=False)
 
     def __post_init__(self):
         if self._node_id is not None:
@@ -160,7 +176,7 @@ class KademliaPeer:
             raise ValueError("invalid udp port")
         if self.tcp_port is not None and not 1 <= self.tcp_port <= 65535:
             raise ValueError("invalid tcp port")
-        if not is_valid_ipv4(self.address):
+        if not is_valid_public_ipv4(self.address, self.allow_localhost):
             raise ValueError(f"invalid ip address: '{self.address}'")
 
     def update_tcp_port(self, tcp_port: int):

lbry/lbry/dht/protocol/iterative_find.py

@@ -153,7 +153,11 @@ class IterativeFinder:
         self._add_active(peer)
         for contact_triple in response.get_close_triples():
             node_id, address, udp_port = contact_triple
+            try:
                 self._add_active(make_kademlia_peer(node_id, address, udp_port))
+            except ValueError:
+                log.warning("misbehaving peer %s:%i returned peer with reserved ip %s:%i", peer.address,
+                            peer.udp_port, address, udp_port)
         self.check_result_ready(response)
 
     async def _send_probe(self, peer: 'KademliaPeer'):
@@ -328,10 +332,17 @@ class IterativeValueFinder(IterativeFinder):
         if not parsed.found:
             return parsed
         already_known = len(self.discovered_peers[peer])
-        self.discovered_peers[peer].update({
+        decoded_peers = set()
-            self.peer_manager.decode_tcp_peer_from_compact_address(compact_addr)
+        for compact_addr in parsed.found_compact_addresses:
-            for compact_addr in parsed.found_compact_addresses
+            try:
-        })
+                decoded_peers.add(self.peer_manager.decode_tcp_peer_from_compact_address(compact_addr))
+            except ValueError:
+                log.warning("misbehaving peer %s:%i returned invalid peer for blob",
+                            peer.address, peer.udp_port)
+                self.peer_manager.report_failure(peer.address, peer.udp_port)
+                parsed.found_compact_addresses.clear()
+                return parsed
+        self.discovered_peers[peer].update(decoded_peers)
         log.debug("probed %s:%i page %i, %i known", peer.address, peer.udp_port, page,
                   already_known + len(parsed.found_compact_addresses))
         if len(self.discovered_peers[peer]) != already_known + len(parsed.found_compact_addresses):

lbry/lbry/extras/daemon/Components.py

@@ -12,6 +12,7 @@ from aioupnp.fault import UPnPError
 
 from lbry import utils
 from lbry.dht.node import Node
+from lbry.dht.peer import is_valid_public_ipv4
 from lbry.dht.blob_announcer import BlobAnnouncer
 from lbry.blob.blob_manager import BlobManager
 from lbry.blob_exchange.server import BlobServer
@@ -385,9 +386,8 @@ class UPnPComponent(Component):
                     log.info("got external ip from UPnP: %s", external_ip)
             except (asyncio.TimeoutError, UPnPError, NotImplementedError):
                 pass
-
+        if external_ip and not is_valid_public_ipv4(external_ip):
-        if external_ip == "0.0.0.0" or (external_ip and external_ip.startswith("192.")):
+            log.warning("UPnP returned a private/reserved ip - %s, checking lbry.com fallback", external_ip)
-            log.warning("unable to get external ip from UPnP, checking lbry.com fallback")
             external_ip = await utils.get_external_ip()
         if self.external_ip and self.external_ip != external_ip:
             log.info("external ip changed from %s to %s", self.external_ip, external_ip)

lbry/lbry/stream/downloader.py

@@ -51,7 +51,7 @@ class StreamDownloader:
         def _delayed_add_fixed_peers():
             self.added_fixed_peers = True
             self.peer_queue.put_nowait([
-                make_kademlia_peer(None, address, None, tcp_port=port + 1)
+                make_kademlia_peer(None, address, None, tcp_port=port + 1, allow_localhost=True)
                 for address, port in addresses
             ])
 

lbry/tests/integration/test_dht.py

@@ -3,6 +3,7 @@ from binascii import hexlify
 
 from lbry.dht import constants
 from lbry.dht.node import Node
+from lbry.dht import peer as dht_peer
 from lbry.dht.peer import PeerManager, make_kademlia_peer
 from torba.testcase import AsyncioTestCase
 
@@ -10,6 +11,8 @@ from torba.testcase import AsyncioTestCase
 class DHTIntegrationTest(AsyncioTestCase):
 
     async def asyncSetUp(self):
+        dht_peer.ALLOW_LOCALHOST = True
+        self.addCleanup(setattr, dht_peer, 'ALLOW_LOCALHOST', False)
         import logging
         logging.getLogger('asyncio').setLevel(logging.ERROR)
         logging.getLogger('lbry.dht').setLevel(logging.WARN)

lbry/tests/unit/blob_exchange/test_transfer_blob.py

@@ -44,7 +44,7 @@ class BlobExchangeTestBase(AsyncioTestCase):
         self.client_storage = SQLiteStorage(self.client_config, os.path.join(self.client_dir, "lbrynet.sqlite"))
         self.client_blob_manager = BlobManager(self.loop, self.client_dir, self.client_storage, self.client_config)
         self.client_peer_manager = PeerManager(self.loop)
-        self.server_from_client = make_kademlia_peer(b'1' * 48, "127.0.0.1", tcp_port=33333)
+        self.server_from_client = make_kademlia_peer(b'1' * 48, "127.0.0.1", tcp_port=33333, allow_localhost=True)
 
         await self.client_storage.open()
         await self.server_storage.open()
@@ -103,7 +103,7 @@ class TestBlobExchange(BlobExchangeTestBase):
         second_client_blob_manager = BlobManager(
             self.loop, second_client_dir, second_client_storage, second_client_conf
         )
-        server_from_second_client = make_kademlia_peer(b'1' * 48, "127.0.0.1", tcp_port=33333)
+        server_from_second_client = make_kademlia_peer(b'1' * 48, "127.0.0.1", tcp_port=33333, allow_localhost=True)
 
         await second_client_storage.open()
         await second_client_blob_manager.setup()
@@ -194,7 +194,7 @@ class TestBlobExchange(BlobExchangeTestBase):
         second_client_blob_manager = BlobManager(
             self.loop, second_client_dir, second_client_storage, second_client_conf
         )
-        server_from_second_client = make_kademlia_peer(b'1' * 48, "127.0.0.1", tcp_port=33333)
+        server_from_second_client = make_kademlia_peer(b'1' * 48, "127.0.0.1", tcp_port=33333, allow_localhost=True)
 
         await second_client_storage.open()
         await second_client_blob_manager.setup()

lbry/tests/unit/dht/protocol/test_kbucket.py

@@ -7,7 +7,7 @@ from lbry.dht import constants
 from torba.testcase import AsyncioTestCase
 
 
-def address_generator(address=(10, 42, 42, 1)):
+def address_generator(address=(1, 2, 3, 4)):
     def increment(addr):
         value = struct.unpack("I", "".join([chr(x) for x in list(addr)[::-1]]).encode())[0] + 1
         new_addr = []

lbry/tests/unit/dht/test_blob_announcer.py

@@ -119,7 +119,7 @@ class TestBlobAnnouncer(AsyncioTestCase):
     async def test_popular_blob(self):
         peer_count = 150
         addresses = [
-            (constants.generate_id(i + 1), socket.inet_ntoa(int(i + 1).to_bytes(length=4, byteorder='big')))
+            (constants.generate_id(i + 1), socket.inet_ntoa(int(i + 0x01000001).to_bytes(length=4, byteorder='big')))
             for i in range(peer_count)
         ]
         blob_hash = b'1' * 48

lbry/tests/unit/dht/test_peer.py

@@ -10,8 +10,8 @@ class PeerTest(AsyncioTestCase):
         self.loop = asyncio.get_event_loop()
         self.peer_manager = PeerManager(self.loop)
         self.node_ids = [generate_id(), generate_id(), generate_id()]
-        self.first_contact = make_kademlia_peer(self.node_ids[1], '127.0.0.1', udp_port=1000)
+        self.first_contact = make_kademlia_peer(self.node_ids[1], '1.0.0.1', udp_port=1000)
-        self.second_contact = make_kademlia_peer(self.node_ids[0], '192.168.0.1', udp_port=1000)
+        self.second_contact = make_kademlia_peer(self.node_ids[0], '1.0.0.2', udp_port=1000)
 
     def test_peer_is_good_unknown_peer(self):
         # Scenario: peer replied, but caller doesn't know the node_id.
@@ -24,21 +24,42 @@ class PeerTest(AsyncioTestCase):
         self.assertIsNone(self.peer_manager.peer_is_good(peer))
 
     def test_make_contact_error_cases(self):
-        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '192.168.1.20', 100000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '1.2.3.4', 100000)
-        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '192.168.1.20.1', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '1.2.3.4.5', 1000)
         self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], 'this is not an ip', 1000)
-        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '192.168.1.20', -1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '1.2.3.4', -1000)
-        self.assertRaises(ValueError, make_kademlia_peer, b'not valid node id', '192.168.1.20', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '1.2.3.4', 0)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '1.2.3.4', 70000)
+        self.assertRaises(ValueError, make_kademlia_peer, b'not valid node id', '1.2.3.4', 1000)
+
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '0.0.0.0', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '10.0.0.1', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '100.64.0.1', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '127.0.0.1', 1000)
+        self.assertIsNotNone(make_kademlia_peer(self.node_ids[1], '127.0.0.1', 1000, allow_localhost=True))
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '192.168.0.1', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '172.16.0.1', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '169.254.1.1', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '192.0.0.2', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '192.0.2.2', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '192.88.99.2', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '198.18.1.1', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '198.51.100.2', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '198.51.100.2', 1000)
+        self.assertRaises(ValueError, make_kademlia_peer, self.node_ids[1], '203.0.113.4', 1000)
+        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)
 
     def test_boolean(self):
         self.assertNotEqual(self.first_contact, self.second_contact)
         self.assertEqual(
-            self.second_contact, make_kademlia_peer(self.node_ids[0], '192.168.0.1', udp_port=1000)
+            self.second_contact, make_kademlia_peer(self.node_ids[0], '1.0.0.2', udp_port=1000)
         )
 
     def test_compact_ip(self):
-        self.assertEqual(self.first_contact.compact_ip(), b'\x7f\x00\x00\x01')
+        self.assertEqual(b'\x01\x00\x00\x01', self.first_contact.compact_ip())
-        self.assertEqual(self.second_contact.compact_ip(), b'\xc0\xa8\x00\x01')
+        self.assertEqual(b'\x01\x00\x00\x02', self.second_contact.compact_ip())
 
 
 @unittest.SkipTest

lbry/tests/unit/stream/test_managed_stream.py

@@ -111,7 +111,7 @@ class TestManagedStream(BlobExchangeTestBase):
         mock_node = mock.Mock(spec=Node)
         q = asyncio.Queue()
 
-        bad_peer = make_kademlia_peer(b'2' * 48, "127.0.0.1", tcp_port=3334)
+        bad_peer = make_kademlia_peer(b'2' * 48, "127.0.0.1", tcp_port=3334, allow_localhost=True)
 
         def _mock_accumulate_peers(q1, q2):
             async def _task():