Revert "Bump protobuf from 3.17.2 to 3.18.3"

This reverts commit 75d64f9dc6.

hub/common.py

@@ -5,6 +5,7 @@ import hmac
 import ipaddress
 import logging
 import logging.handlers
+import socket
 import typing
 import collections
 from bisect import insort_right
@@ -153,6 +154,38 @@ def protocol_version(client_req, min_tuple, max_tuple):
     return result, client_min
 
 
+async def resolve_host(url: str, port: int, proto: str,
+                       family: int = socket.AF_INET, all_results: bool = False) \
+        -> typing.Union[str, typing.List[str]]:
+    if proto not in ['udp', 'tcp']:
+        raise Exception("invalid protocol")
+    try:
+        if ipaddress.ip_address(url):
+            return [url] if all_results else url
+    except ValueError:
+        pass
+    loop = asyncio.get_running_loop()
+    records = await loop.getaddrinfo(
+        url, port,
+        proto=socket.IPPROTO_TCP if proto == 'tcp' else socket.IPPROTO_UDP,
+        type=socket.SOCK_STREAM if proto == 'tcp' else socket.SOCK_DGRAM,
+        family=family,
+    )
+    def addr_not_ipv4_mapped(rec):
+        _, _, _, _, sockaddr = rec
+        ipaddr = ipaddress.ip_address(sockaddr[0])
+        return ipaddr.version != 6 or not ipaddr.ipv4_mapped
+    records = filter(addr_not_ipv4_mapped, records)
+    results = [sockaddr[0] for fam, type, prot, canonname, sockaddr in records]
+    if not results and not all_results:
+        raise socket.gaierror(
+            socket.EAI_ADDRFAMILY,
+            'The specified network host does not have any network '
+            'addresses in the requested address family'
+        )
+    return results if all_results else results[0]
+
+
 class LRUCacheWithMetrics:
     __slots__ = [
         'capacity',
@@ -577,10 +610,10 @@ IPV4_TO_6_RELAY_SUBNET = ipaddress.ip_network('192.88.99.0/24')
 def is_valid_public_ipv4(address, allow_localhost: bool = False, allow_lan: bool = False):
     try:
         parsed_ip = ipaddress.ip_address(address)
-        if parsed_ip.is_loopback and allow_localhost:
-            return True
-        if allow_lan and parsed_ip.is_private:
-            return True
+        if parsed_ip.is_loopback:
+            return allow_localhost
+        if parsed_ip.is_private:
+            return allow_lan
         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)):
             return False
@@ -590,6 +623,22 @@ def is_valid_public_ipv4(address, allow_localhost: bool = False, allow_lan: bool
     except (ipaddress.AddressValueError, ValueError):
         return False
 
+def is_valid_public_ipv6(address, allow_localhost: bool = False, allow_lan: bool = False):
+    try:
+        parsed_ip = ipaddress.ip_address(address)
+        if parsed_ip.is_loopback:
+            return allow_localhost
+        if parsed_ip.is_private:
+            return allow_lan
+        return not any((parsed_ip.version != 6, 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.ipv4_mapped))
+    except (ipaddress.AddressValueError, ValueError):
+        return False
+
+def is_valid_public_ip(address, **kwargs):
+    return is_valid_public_ipv6(address, **kwargs) or is_valid_public_ipv4(address, **kwargs)
 
 def sha256(x):
     """Simple wrapper of hashlib sha256."""

hub/env.py

@@ -117,10 +117,6 @@ class Env:
         result = [part.strip() for part in host.split(',')]
         if len(result) == 1:
             result = result[0]
-        if result == 'localhost':
-            # 'localhost' resolves to ::1 (ipv6) on many systems, which fails on default setup of
-            # docker, using 127.0.0.1 instead forces ipv4
-            result = '127.0.0.1'
         return result
 
     def sane_max_sessions(self):

hub/herald/service.py

@@ -1,3 +1,4 @@
+import errno
 import time
 import typing
 import asyncio
@@ -170,10 +171,20 @@ class HubServerService(BlockchainReaderService):
 
     async def start_status_server(self):
         if self.env.udp_port and int(self.env.udp_port):
-            await self.status_server.start(
-                0, bytes.fromhex(self.env.coin.GENESIS_HASH)[::-1], self.env.country,
-                self.env.host, self.env.udp_port, self.env.allow_lan_udp
-            )
+            hosts = self.env.cs_host()
+            started = False
+            while not started:
+                try:
+                    await self.status_server.start(
+                        0, bytes.fromhex(self.env.coin.GENESIS_HASH)[::-1], self.env.country,
+                        hosts, self.env.udp_port, self.env.allow_lan_udp
+                    )
+                    started = True
+                except OSError as e:
+                    if e.errno is errno.EADDRINUSE:
+                        await asyncio.sleep(3)
+                        continue
+                    raise
 
     def _iter_start_tasks(self):
         yield self.start_status_server()

hub/herald/session.py

@@ -271,7 +271,8 @@ class SessionManager:
                                   f'{host}:{port:d} : {e!r}')
             raise
         else:
-            self.logger.info(f'{kind} server listening on {host}:{port:d}')
+            for s in self.servers[kind].sockets:
+                self.logger.info(f'{kind} server listening on {s.getsockname()[:2]}')
 
     async def _start_external_servers(self):
         """Start listening on TCP and SSL ports, but only if the respective

hub/herald/udp.py

@@ -1,10 +1,18 @@
 import asyncio
+import ipaddress
+import socket
 import struct
 from time import perf_counter
 import logging
-from typing import Optional, Tuple, NamedTuple
+from typing import Optional, Tuple, NamedTuple, List, Union
 from hub.schema.attrs import country_str_to_int, country_int_to_str
-from hub.common import LRUCache, is_valid_public_ipv4
+from hub.common import (
+    LRUCache,
+    resolve_host,
+    is_valid_public_ip,
+    is_valid_public_ipv4,
+    is_valid_public_ipv6,
+)
 
 
 log = logging.getLogger(__name__)
@@ -36,48 +44,75 @@ class SPVPing(NamedTuple):
         return decoded
 
 
-PONG_ENCODING = b'!BBL32s4sH'
-
+PONG_ENCODING_PRE = b'!BBL32s'
+PONG_ENCODING_POST = b'!H'
 
 class SPVPong(NamedTuple):
     protocol_version: int
     flags: int
     height: int
     tip: bytes
-    source_address_raw: bytes
+    ipaddr: Union[ipaddress.IPv4Address, ipaddress.IPv6Address]
     country: int
 
+    FLAG_AVAILABLE = 0b00000001
+    FLAG_IPV6 = 0b00000010
+
     def encode(self):
-        return struct.pack(PONG_ENCODING, *self)
+        return (struct.pack(PONG_ENCODING_PRE, self.protocol_version, self.flags, self.height, self.tip) +
+                self.encode_address(self.ipaddr) +
+                struct.pack(PONG_ENCODING_POST, self.country))
 
     @staticmethod
-    def encode_address(address: str):
-        return bytes(int(b) for b in address.split("."))
+    def encode_address(address: Union[str, ipaddress.IPv4Address, ipaddress.IPv6Address]):
+        if not isinstance(address, (ipaddress.IPv4Address, ipaddress.IPv6Address)):
+            address = ipaddress.ip_address(address)
+        return address.packed
 
     @classmethod
     def make(cls, flags: int, height: int, tip: bytes, source_address: str, country: str) -> bytes:
+        ipaddr = ipaddress.ip_address(source_address)
+        flags = (flags | cls.FLAG_IPV6) if ipaddr.version == 6 else (flags & ~cls.FLAG_IPV6)
         return SPVPong(
             PROTOCOL_VERSION, flags, height, tip,
-            cls.encode_address(source_address),
+            ipaddr,
             country_str_to_int(country)
-        ).encode()
+        )
 
     @classmethod
     def make_sans_source_address(cls, flags: int, height: int, tip: bytes, country: str) -> Tuple[bytes, bytes]:
         pong = cls.make(flags, height, tip, '0.0.0.0', country)
-        return pong[:38], pong[42:]
+        pong = pong.encode()
+        return pong[0:1], pong[2:38], pong[42:]
 
     @classmethod
     def decode(cls, packet: bytes):
-        return cls(*struct.unpack(PONG_ENCODING, packet[:44]))
+        offset = 0
+        protocol_version, flags, height, tip = struct.unpack(PONG_ENCODING_PRE, packet[offset:offset+38])
+        offset += 38
+        if flags & cls.FLAG_IPV6:
+            addr_len = ipaddress.IPV6LENGTH // 8
+            ipaddr = ipaddress.ip_address(packet[offset:offset+addr_len])
+            offset += addr_len
+        else:
+            addr_len = ipaddress.IPV4LENGTH // 8
+            ipaddr = ipaddress.ip_address(packet[offset:offset+addr_len])
+            offset += addr_len
+        country, = struct.unpack(PONG_ENCODING_POST, packet[offset:offset+2])
+        offset += 2
+        return cls(protocol_version, flags, height, tip, ipaddr, country)
 
     @property
     def available(self) -> bool:
-        return (self.flags & 0b00000001) > 0
+        return (self.flags & self.FLAG_AVAILABLE) > 0
+
+    @property
+    def ipv6(self) -> bool:
+        return (self.flags & self.FLAG_IPV6) > 0
 
     @property
     def ip_address(self) -> str:
-        return ".".join(map(str, self.source_address_raw))
+        return self.ipaddr.compressed
 
     @property
     def country_name(self):
@@ -94,7 +129,8 @@ class SPVServerStatusProtocol(asyncio.DatagramProtocol):
     def __init__(
         self, height: int, tip: bytes, country: str,
         throttle_cache_size: int = 1024, throttle_reqs_per_sec: int = 10,
-        allow_localhost: bool = False, allow_lan: bool = False
+        allow_localhost: bool = False, allow_lan: bool = False,
+        is_valid_ip = is_valid_public_ip,
     ):
         super().__init__()
         self.transport: Optional[asyncio.transports.DatagramTransport] = None
@@ -102,26 +138,27 @@ class SPVServerStatusProtocol(asyncio.DatagramProtocol):
         self._tip = tip
         self._flags = 0
         self._country = country
-        self._left_cache = self._right_cache = None
+        self._cache0 = self._cache1 = self.cache2 = None
         self.update_cached_response()
         self._throttle = LRUCache(throttle_cache_size)
         self._should_log = LRUCache(throttle_cache_size)
         self._min_delay = 1 / throttle_reqs_per_sec
         self._allow_localhost = allow_localhost
         self._allow_lan = allow_lan
+        self._is_valid_ip = is_valid_ip
         self.closed = asyncio.Event()
 
     def update_cached_response(self):
-        self._left_cache, self._right_cache = SPVPong.make_sans_source_address(
+        self._cache0, self._cache1, self._cache2 = SPVPong.make_sans_source_address(
             self._flags, max(0, self._height), self._tip, self._country
         )
 
     def set_unavailable(self):
-        self._flags &= 0b11111110
+        self._flags &= ~SPVPong.FLAG_AVAILABLE
         self.update_cached_response()
 
     def set_available(self):
-        self._flags |= 0b00000001
+        self._flags |= SPVPong.FLAG_AVAILABLE
         self.update_cached_response()
 
     def set_height(self, height: int, tip: bytes):
@@ -141,17 +178,25 @@ class SPVServerStatusProtocol(asyncio.DatagramProtocol):
         return False
 
     def make_pong(self, host):
-        return self._left_cache + SPVPong.encode_address(host) + self._right_cache
+        ipaddr = ipaddress.ip_address(host)
+        if ipaddr.version == 6:
+            flags = self._flags | SPVPong.FLAG_IPV6
+        else:
+            flags = self._flags & ~SPVPong.FLAG_IPV6
+        return (self._cache0 + flags.to_bytes(1, 'big') +
+                self._cache1 + SPVPong.encode_address(ipaddr) +
+                self._cache2)
 
-    def datagram_received(self, data: bytes, addr: Tuple[str, int]):
+    def datagram_received(self, data: bytes, addr: Union[Tuple[str, int], Tuple[str, int, int, int]]):
         if self.should_throttle(addr[0]):
+            # print(f"throttled: {addr}")
             return
         try:
             SPVPing.decode(data)
         except (ValueError, struct.error, AttributeError, TypeError):
             # log.exception("derp")
             return
-        if addr[1] >= 1024 and is_valid_public_ipv4(
+        if addr[1] >= 1024 and self._is_valid_ip(
                 addr[0], allow_localhost=self._allow_localhost, allow_lan=self._allow_lan):
             self.transport.sendto(self.make_pong(addr[0]), addr)
         else:
@@ -174,39 +219,78 @@ class SPVServerStatusProtocol(asyncio.DatagramProtocol):
 
 class StatusServer:
     def __init__(self):
-        self._protocol: Optional[SPVServerStatusProtocol] = None
+        self._protocols: List[SPVServerStatusProtocol] = []
 
-    async def start(self, height: int, tip: bytes, country: str, interface: str, port: int, allow_lan: bool = False):
-        if self.is_running:
-            return
-        loop = asyncio.get_event_loop()
-        interface = interface if interface.lower() != 'localhost' else '127.0.0.1'
-        self._protocol = SPVServerStatusProtocol(
-            height, tip, country, allow_localhost=interface == '127.0.0.1', allow_lan=allow_lan
-        )
-        await loop.create_datagram_endpoint(lambda: self._protocol, (interface, port))
-        log.info("started udp status server on %s:%i", interface, port)
+    async def _start(self, height: int, tip: bytes, country: str, addr: str, port: int, allow_lan: bool = False):
+        ipaddr = ipaddress.ip_address(addr)
+        if ipaddr.version == 4:
+            proto = SPVServerStatusProtocol(
+                height, tip, country,
+                allow_localhost=ipaddr.is_loopback or ipaddr.is_unspecified,
+                allow_lan=allow_lan,
+                is_valid_ip=is_valid_public_ipv4,
+            )
+            loop = asyncio.get_event_loop()
+            await loop.create_datagram_endpoint(lambda: proto, (ipaddr.compressed, port), family=socket.AF_INET)
+        elif ipaddr.version == 6:
+            proto = SPVServerStatusProtocol(
+                height, tip, country,
+                allow_localhost=ipaddr.is_loopback or ipaddr.is_unspecified,
+                allow_lan=allow_lan,
+                is_valid_ip=is_valid_public_ipv6,
+            )
+            # Because dualstack / IPv4 mapped address behavior on an IPv6 socket
+            # differs based on system config, create the socket with IPV6_V6ONLY.
+            # This disables the IPv4 mapped feature, so we don't need to consider
+            # when an IPv6 socket may interfere with IPv4 binding / traffic.
+            sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
+            sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
+            sock.bind((ipaddr.compressed, port))
+            loop = asyncio.get_event_loop()
+            await loop.create_datagram_endpoint(lambda: proto, sock=sock)
+        else:
+            raise ValueError(f'unexpected IP address version {ipaddr.version}')
+        log.info("started udp%i status server on %s", ipaddr.version, proto.transport.get_extra_info('sockname')[:2])
+        self._protocols.append(proto)
+
+    async def start(self, height: int, tip: bytes, country: str, hosts: List[str], port: int, allow_lan: bool = False):
+        if not isinstance(hosts, list):
+            hosts = [hosts]
+        try:
+            for host in hosts:
+                addr = None
+                if not host:
+                    resolved = ['::', '0.0.0.0'] # unspecified address
+                else:
+                    resolved = await resolve_host(host, port, 'udp', family=socket.AF_UNSPEC, all_results=True)
+                for addr in resolved:
+                    await self._start(height, tip, country, addr, port, allow_lan)
+        except Exception as e:
+            if not isinstance(e, asyncio.CancelledError):
+                log.error("UDP status server failed to listen on (%s:%i) : %s", addr or host, port, e)
+            await self.stop()
+            raise
 
     async def stop(self):
-        if self.is_running:
-            await self._protocol.close()
-            self._protocol = None
+        for proto in self._protocols:
+            await proto.close()
+        self._protocols.clear()
 
     @property
     def is_running(self):
-        return self._protocol is not None
+        return self._protocols
 
     def set_unavailable(self):
-        if self.is_running:
-            self._protocol.set_unavailable()
+        for proto in self._protocols:
+            proto.set_unavailable()
 
     def set_available(self):
-        if self.is_running:
-            self._protocol.set_available()
+        for proto in self._protocols:
+            proto.set_available()
 
     def set_height(self, height: int, tip: bytes):
-        if self.is_running:
-            self._protocol.set_height(height, tip)
+        for proto in self._protocols:
+            proto.set_height(height, tip)
 
 
 class SPVStatusClientProtocol(asyncio.DatagramProtocol):
@@ -217,9 +301,14 @@ class SPVStatusClientProtocol(asyncio.DatagramProtocol):
         self.responses = responses
         self._ping_packet = SPVPing.make()
 
-    def datagram_received(self, data: bytes, addr: Tuple[str, int]):
+    def datagram_received(self, data: bytes, addr: Union[Tuple[str, int], Tuple[str, int, int, int]]):
         try:
-            self.responses.put_nowait(((addr, perf_counter()), SPVPong.decode(data)))
+            if len(addr) > 2: # IPv6 with possible mapped IPv4
+                ipaddr = ipaddress.ip_address(addr[0])
+                if ipaddr.ipv4_mapped:
+                    # mapped IPv4 address identified
+                    addr = (ipaddr.ipv4_mapped.compressed, addr[1])
+            self.responses.put_nowait(((addr[:2], perf_counter()), SPVPong.decode(data)))
         except (ValueError, struct.error, AttributeError, TypeError, RuntimeError):
             return
 
@@ -230,7 +319,7 @@ class SPVStatusClientProtocol(asyncio.DatagramProtocol):
         self.transport = None
         log.info("closed udp spv server selection client")
 
-    def ping(self, server: Tuple[str, int]):
+    def ping(self, server: Union[Tuple[str, int], Tuple[str, int, int, int]]):
         self.transport.sendto(self._ping_packet, server)
 
     def close(self):

setup.py

@@ -33,7 +33,7 @@ setup(
         'certifi>=2021.10.08',
         'colorama==0.3.7',
         'cffi==1.13.2',
-        'protobuf==3.18.3',
+        'protobuf==3.17.2',
         'msgpack==0.6.1',
         'prometheus_client==0.7.1',
         'coincurve==15.0.0',