Madiator2011/lbry-sdk
1
0
Fork 0
forked from LBRYCommunity/lbry-sdk
Code Pull requests Activity

Merge branch 'endes123321-master'

Browse source
This commit is contained in:
Jack Robison 2020-06-02 15:59:53 -04:00
commit f975ea99cb
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
2 changed files with 34 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
lbry/dht/peer.py
View file

@ -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")),
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:
except (ipaddress.AddressValueError, ValueError):
return False

27
tests/unit/dht/test_peer.py
View file

@ -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)