Ignore invalid IPv6 addresses

node.py

@@ -36,6 +36,25 @@ class Node:
             self.partners = []
             return heartbeats, partners
 
+def is_loopback(ipv6):
+    return ipv6 == 15 * b"\0" + b"\x01"
+
+def is_unspecified(ipv6):
+    return ipv6 == 16 * b"\0"
+
+def is_mapped_ipv4(ipv6):
+    return ipv6[0:12] == 10 * b"\0" + 2 * b"\xff"
+
+def is_local_unicast(ipv6):
+    return (ipv6[0] & 0xfe) == 0xfc
+
+def is_valid_address(ipv6):
+    ipv6 = socket.inet_pton(socket.AF_INET6, ipv6)
+    return not is_loopback(ipv6) and \
+        not is_unspecified(ipv6) and \
+        not is_mapped_ipv4(ipv6) and \
+        not is_local_unicast(ipv6)
+
 def describe(ipv6, port):
     if port == DEFAULT_PORT:
         return ipv6
@@ -122,6 +141,9 @@ def heartbeat(node):
 
         # adding new peers that contacted me
         for new_peer in (heartbeats - partners):
+            if not is_valid_address(new_peer[0]):
+                log(f"Ignoring this address: {new_peer[0]}")
+                continue
             peer = Peer(new_peer[0], new_peer[1], 10)
             log(f"Adding peer {peer} to the list (It connected on its own)")
             node.peers.append(peer)
@@ -129,6 +151,9 @@ def heartbeat(node):
 
         # adding partners
         for new_peer in partners:
+            if not is_valid_address(new_peer[0]):
+                log(f"Ignoring this address: {new_peer[0]}")
+                continue
             peer = Peer(new_peer[0], new_peer[1], 3)
             log(f"Adding peer {peer} to the list (Got it as \"partner\" from another peer)")
             node.peers.append(peer)