diff --git a/hub/common.py b/hub/common.py
index 4e150ac..e6645dc 100644
--- a/hub/common.py
+++ b/hub/common.py
@@ -5,6 +5,7 @@ import hmac
import ipaddress
import logging
import logging.handlers
+import socket
import typing
import collections
from ctypes import cast, memmove, POINTER, c_void_p
@@ -155,6 +156,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',
@@ -579,11 +612,13 @@ 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 any((parsed_ip.version != 4, parsed_ip.is_unspecified, parsed_ip.is_link_local, parsed_ip.is_loopback,
+ if parsed_ip.version != 4:
+ return False
+ if parsed_ip.is_loopback:
+ return allow_localhost
+ if parsed_ip.is_private:
+ return allow_lan
+ if any((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
else:
@@ -592,6 +627,23 @@ 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.version != 6:
+ return False
+ if parsed_ip.is_loopback:
+ return allow_localhost
+ if parsed_ip.is_private:
+ return allow_lan
+ return not any((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."""
diff --git a/hub/env.py b/hub/env.py
index d7edbdc..645ca97 100644
--- a/hub/env.py
+++ b/hub/env.py
@@ -116,10 +116,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):
diff --git a/hub/herald/service.py b/hub/herald/service.py
index 8d504b7..ce9a883 100644
--- a/hub/herald/service.py
+++ b/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()
diff --git a/hub/herald/session.py b/hub/herald/session.py
index 56b33d2..d981dfa 100644
--- a/hub/herald/session.py
+++ b/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
diff --git a/hub/herald/udp.py b/hub/herald/udp.py
index af83d0c..6fd388c 100644
--- a/hub/herald/udp.py
+++ b/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):
diff --git a/setup.py b/setup.py
index 71a198e..7c65abc 100644
--- a/setup.py
+++ b/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',