76f69676d3
Some regtest tests depend on manual fee injection to simulate certain mempool conditions (e.g. lnwatcher_waits_until_fees_go_down). This is done by manually injecting fee estimates into the `Network` object using the `test_inject_fee_etas` cli command. However it can still happen that the Network automatically updates its fee estimates from the connected electrum server in the time between injecting the fee and the actual tested logic making decisions based on the fee. This causes the test to fail sometimes. By setting the `test_disable_automatic_fee_eta_update` true the Network will stop automatically updating the fee estimates and the test will behave as expected.
1373 lines
56 KiB
Python
1373 lines
56 KiB
Python
# Electrum - Lightweight Bitcoin Client
|
|
# Copyright (c) 2011-2016 Thomas Voegtlin
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person
|
|
# obtaining a copy of this software and associated documentation files
|
|
# (the "Software"), to deal in the Software without restriction,
|
|
# including without limitation the rights to use, copy, modify, merge,
|
|
# publish, distribute, sublicense, and/or sell copies of the Software,
|
|
# and to permit persons to whom the Software is furnished to do so,
|
|
# subject to the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be
|
|
# included in all copies or substantial portions of the Software.
|
|
#
|
|
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
|
|
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
|
|
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
|
|
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
|
|
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
|
|
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
|
|
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
|
# SOFTWARE.
|
|
import asyncio
|
|
import time
|
|
import os
|
|
import random
|
|
import re
|
|
from collections import defaultdict
|
|
import threading
|
|
import json
|
|
from typing import (
|
|
NamedTuple, Optional, Sequence, List, Dict, Tuple, TYPE_CHECKING, Iterable, Set, Any, TypeVar,
|
|
Callable
|
|
)
|
|
import copy
|
|
import functools
|
|
from enum import IntEnum
|
|
from contextlib import nullcontext
|
|
|
|
import aiorpcx
|
|
from aiorpcx import ignore_after, NetAddress
|
|
from aiohttp import ClientResponse
|
|
|
|
from . import util
|
|
from .util import (
|
|
log_exceptions, ignore_exceptions, OldTaskGroup, make_aiohttp_session,
|
|
NetworkRetryManager, error_text_str_to_safe_str, detect_tor_socks_proxy
|
|
)
|
|
from . import constants
|
|
from . import blockchain
|
|
from . import dns_hacks
|
|
from .transaction import Transaction
|
|
from .blockchain import Blockchain
|
|
from .interface import (
|
|
Interface, PREFERRED_NETWORK_PROTOCOL, RequestTimedOut, NetworkTimeout, BUCKET_NAME_OF_ONION_SERVERS,
|
|
NetworkException, RequestCorrupted, ServerAddr, TxBroadcastError,
|
|
)
|
|
from .version import PROTOCOL_VERSION
|
|
from .i18n import _
|
|
from .logging import get_logger, Logger
|
|
from .fee_policy import FeeHistogram, FeeTimeEstimates, FEE_ETA_TARGETS
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Coroutine
|
|
from .channel_db import ChannelDB
|
|
from .lnrouter import LNPathFinder
|
|
from .lnworker import LNGossip
|
|
from .daemon import Daemon
|
|
from .simple_config import SimpleConfig
|
|
|
|
|
|
_logger = get_logger(__name__)
|
|
|
|
|
|
NUM_TARGET_CONNECTED_SERVERS = 10
|
|
NUM_STICKY_SERVERS = 4
|
|
NUM_RECENT_SERVERS = 20
|
|
|
|
T = TypeVar('T')
|
|
|
|
|
|
class ConnectionState(IntEnum):
|
|
DISCONNECTED = 0
|
|
CONNECTING = 1
|
|
CONNECTED = 2
|
|
|
|
|
|
def parse_servers(result: Sequence[Tuple[str, str, List[str]]]) -> Dict[str, dict]:
|
|
"""Convert servers list (from protocol method "server.peers.subscribe") into dict format.
|
|
Also validate values, such as IP addresses and ports.
|
|
"""
|
|
servers = {}
|
|
for item in result:
|
|
host = item[1]
|
|
out = {}
|
|
version = None
|
|
pruning_level = '-'
|
|
if len(item) > 2:
|
|
for v in item[2]:
|
|
if re.match(r"[st]\d*", v):
|
|
protocol, port = v[0], v[1:]
|
|
if port == '': port = constants.net.DEFAULT_PORTS[protocol]
|
|
ServerAddr(host, port, protocol=protocol) # check if raises
|
|
out[protocol] = port
|
|
elif re.match("v(.?)+", v):
|
|
version = v[1:]
|
|
elif re.match(r"p\d*", v):
|
|
pruning_level = v[1:]
|
|
if pruning_level == '': pruning_level = '0'
|
|
if out:
|
|
out['pruning'] = pruning_level
|
|
out['version'] = version
|
|
servers[host] = out
|
|
return servers
|
|
|
|
|
|
def filter_version(servers):
|
|
def is_recent(version):
|
|
try:
|
|
return util.versiontuple(version) >= util.versiontuple(PROTOCOL_VERSION)
|
|
except Exception as e:
|
|
return False
|
|
return {k: v for k, v in servers.items() if is_recent(v.get('version'))}
|
|
|
|
|
|
def filter_noonion(servers):
|
|
return {k: v for k, v in servers.items() if not k.endswith('.onion')}
|
|
|
|
|
|
def filter_protocol(hostmap, *, allowed_protocols: Iterable[str] = None) -> Sequence[ServerAddr]:
|
|
"""Filters the hostmap for those implementing protocol."""
|
|
if allowed_protocols is None:
|
|
allowed_protocols = {PREFERRED_NETWORK_PROTOCOL}
|
|
eligible = []
|
|
for host, portmap in hostmap.items():
|
|
for protocol in allowed_protocols:
|
|
port = portmap.get(protocol)
|
|
if port:
|
|
eligible.append(ServerAddr(host, port, protocol=protocol))
|
|
return eligible
|
|
|
|
|
|
def pick_random_server(hostmap=None, *, allowed_protocols: Iterable[str],
|
|
exclude_set: Set[ServerAddr] = None) -> Optional[ServerAddr]:
|
|
if hostmap is None:
|
|
hostmap = constants.net.DEFAULT_SERVERS
|
|
if exclude_set is None:
|
|
exclude_set = set()
|
|
servers = set(filter_protocol(hostmap, allowed_protocols=allowed_protocols))
|
|
eligible = list(servers - exclude_set)
|
|
return random.choice(eligible) if eligible else None
|
|
|
|
|
|
def is_valid_port(ps: str):
|
|
try:
|
|
return 0 < int(ps) < 65535
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def is_valid_host(ph: str):
|
|
try:
|
|
NetAddress(ph, '1')
|
|
except ValueError:
|
|
return False
|
|
return True
|
|
|
|
|
|
class ProxySettings:
|
|
MODES = ['socks4', 'socks5']
|
|
|
|
probe_fut = None
|
|
|
|
def __init__(self):
|
|
self.enabled = False
|
|
self.mode = 'socks5'
|
|
self.host = ''
|
|
self.port = ''
|
|
self.user = None
|
|
self.password = None
|
|
|
|
def set_defaults(self):
|
|
self.__init__() # call __init__ for default values
|
|
|
|
def serialize_proxy_cfgstr(self):
|
|
return ':'.join([self.mode, self.host, self.port])
|
|
|
|
def deserialize_proxy_cfgstr(self, s: Optional[str], user: str = None, password: str = None) -> None:
|
|
if s is None or (isinstance(s, str) and s.lower() == 'none'):
|
|
self.set_defaults()
|
|
self.user = user
|
|
self.password = password
|
|
return
|
|
|
|
if not isinstance(s, str):
|
|
raise ValueError('proxy config not a string')
|
|
|
|
args = s.split(':')
|
|
if args[0] in ProxySettings.MODES:
|
|
self.mode = args[0]
|
|
args = args[1:]
|
|
|
|
# detect migrate from old settings
|
|
if len(args) == 4 and is_valid_host(args[0]) and is_valid_port(args[1]): # host:port:user:pass,
|
|
self.host = args[0]
|
|
self.port = args[1]
|
|
self.user = args[2]
|
|
self.password = args[3]
|
|
else:
|
|
self.host = ':'.join(args[:-1])
|
|
self.port = args[-1]
|
|
self.user = user
|
|
self.password = password
|
|
|
|
if not is_valid_host(self.host) or not is_valid_port(self.port):
|
|
self.enabled = False
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'enabled': self.enabled,
|
|
'mode': self.mode,
|
|
'host': self.host,
|
|
'port': self.port,
|
|
'user': self.user,
|
|
'password': self.password
|
|
}
|
|
|
|
@classmethod
|
|
def from_config(cls, config: 'SimpleConfig') -> 'ProxySettings':
|
|
proxy = ProxySettings()
|
|
proxy.deserialize_proxy_cfgstr(
|
|
config.NETWORK_PROXY, config.NETWORK_PROXY_USER, config.NETWORK_PROXY_PASSWORD
|
|
)
|
|
proxy.enabled = config.NETWORK_PROXY_ENABLED
|
|
return proxy
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: dict) -> 'ProxySettings':
|
|
proxy = ProxySettings()
|
|
proxy.enabled = d.get('enabled', proxy.enabled)
|
|
proxy.mode = d.get('mode', proxy.mode)
|
|
proxy.host = d.get('host', proxy.host)
|
|
proxy.port = d.get('port', proxy.port)
|
|
proxy.user = d.get('user', proxy.user)
|
|
proxy.password = d.get('password', proxy.password)
|
|
return proxy
|
|
|
|
@classmethod
|
|
def probe_tor(cls, on_finished: Callable[[str | None, int | None], None]):
|
|
async def detect_task(finished: Callable[[str | None, int | None], None]):
|
|
try:
|
|
net_addr = await detect_tor_socks_proxy()
|
|
if net_addr is None:
|
|
finished('', -1)
|
|
else:
|
|
host = net_addr[0]
|
|
port = net_addr[1]
|
|
finished(host, port)
|
|
finally:
|
|
cls.probe_fut = None
|
|
|
|
if cls.probe_fut: # one probe at a time
|
|
return
|
|
cls.probe_fut = asyncio.run_coroutine_threadsafe(detect_task(on_finished), util.get_asyncio_loop())
|
|
|
|
def __eq__(self, other):
|
|
return self.enabled == other.enabled \
|
|
and self.mode == other.mode \
|
|
and self.host == other.host \
|
|
and self.port == other.port \
|
|
and self.user == other.user \
|
|
and self.password == other.password
|
|
|
|
def __str__(self):
|
|
return f'{self.enabled=} {self.mode=} {self.host=} {self.port=} {self.user=}'
|
|
|
|
|
|
class NetworkParameters(NamedTuple):
|
|
server: ServerAddr
|
|
proxy: ProxySettings
|
|
auto_connect: bool
|
|
oneserver: bool = False
|
|
|
|
|
|
class BestEffortRequestFailed(NetworkException): pass
|
|
|
|
|
|
class UntrustedServerReturnedError(NetworkException):
|
|
def __init__(self, *, original_exception):
|
|
self.original_exception = original_exception
|
|
|
|
def get_message_for_gui(self) -> str:
|
|
return str(self)
|
|
|
|
def get_untrusted_message(self) -> str:
|
|
e = self.original_exception
|
|
return (f"<UntrustedServerReturnedError "
|
|
f"[DO NOT TRUST THIS MESSAGE] original_exception: {error_text_str_to_safe_str(repr(e))}>")
|
|
|
|
def __str__(self):
|
|
# We should not show the untrusted text from self.original_exception,
|
|
# to avoid accidentally showing it in the GUI.
|
|
return _("The server returned an error.")
|
|
|
|
def __repr__(self):
|
|
# We should not show the untrusted text from self.original_exception,
|
|
# to avoid accidentally showing it in the GUI.
|
|
return f"<UntrustedServerReturnedError {str(self)!r}>"
|
|
|
|
|
|
_INSTANCE = None
|
|
|
|
|
|
class Network(Logger, NetworkRetryManager[ServerAddr]):
|
|
"""The Network class manages a set of connections to remote electrum
|
|
servers, each connected socket is handled by an Interface() object.
|
|
"""
|
|
|
|
taskgroup: Optional[OldTaskGroup]
|
|
interface: Optional[Interface]
|
|
interfaces: Dict[ServerAddr, Interface]
|
|
_connecting_ifaces: Set[ServerAddr]
|
|
_closing_ifaces: Set[ServerAddr]
|
|
default_server: ServerAddr
|
|
_recent_servers: List[ServerAddr]
|
|
|
|
channel_db: Optional['ChannelDB'] = None
|
|
lngossip: Optional['LNGossip'] = None
|
|
path_finder: Optional['LNPathFinder'] = None
|
|
|
|
def __init__(self, config: 'SimpleConfig', *, daemon: 'Daemon' = None):
|
|
global _INSTANCE
|
|
assert _INSTANCE is None, "Network is a singleton!"
|
|
_INSTANCE = self
|
|
|
|
Logger.__init__(self)
|
|
NetworkRetryManager.__init__(
|
|
self,
|
|
max_retry_delay_normal=600,
|
|
init_retry_delay_normal=15,
|
|
max_retry_delay_urgent=10,
|
|
init_retry_delay_urgent=1,
|
|
)
|
|
|
|
self.asyncio_loop = util.get_asyncio_loop()
|
|
assert self.asyncio_loop.is_running(), "event loop not running"
|
|
|
|
self.config = config
|
|
self.daemon = daemon
|
|
|
|
blockchain.read_blockchains(self.config)
|
|
blockchain.init_headers_file_for_best_chain()
|
|
self.logger.info(f"blockchains {list(map(lambda b: b.forkpoint, blockchain.blockchains.values()))}")
|
|
self._blockchain_preferred_block = self.config.BLOCKCHAIN_PREFERRED_BLOCK # type: Dict[str, Any]
|
|
if self._blockchain_preferred_block is None:
|
|
self._set_preferred_chain(None)
|
|
self._blockchain = blockchain.get_best_chain()
|
|
|
|
self._allowed_protocols = {PREFERRED_NETWORK_PROTOCOL}
|
|
|
|
self.proxy = ProxySettings()
|
|
self.is_proxy_tor = None # type: Optional[bool] # tri-state. None means unknown.
|
|
self._init_parameters_from_config()
|
|
|
|
self.taskgroup = None
|
|
|
|
# locks
|
|
self.restart_lock = asyncio.Lock()
|
|
self.bhi_lock = asyncio.Lock()
|
|
self.recent_servers_lock = threading.RLock() # <- re-entrant
|
|
self.interfaces_lock = threading.Lock() # for mutating/iterating self.interfaces
|
|
|
|
self.server_peers = {} # returned by interface (servers that the main interface knows about)
|
|
self._recent_servers = self._read_recent_servers() # note: needs self.recent_servers_lock
|
|
|
|
self.banner = ''
|
|
self.donation_address = ''
|
|
self.relay_fee = None # type: Optional[int]
|
|
|
|
dir_path = os.path.join(self.config.path, 'certs')
|
|
util.make_dir(dir_path)
|
|
|
|
# the main server we are currently communicating with
|
|
self.interface = None
|
|
self.default_server_changed_event = asyncio.Event()
|
|
# Set of servers we have an ongoing connection with.
|
|
# For any ServerAddr, at most one corresponding Interface object
|
|
# can exist at any given time. Depending on the state of that Interface,
|
|
# the ServerAddr can be found in one of the following sets.
|
|
# Note: during a transition, the ServerAddr can appear in two sets momentarily.
|
|
self._connecting_ifaces = set()
|
|
self.interfaces = {} # these are the ifaces in "initialised and usable" state
|
|
self._closing_ifaces = set()
|
|
|
|
# Dump network messages (all interfaces). Set at runtime from the console.
|
|
self.debug = False
|
|
|
|
self._set_status(ConnectionState.DISCONNECTED)
|
|
self._has_ever_managed_to_connect_to_server = False
|
|
self._was_started = False
|
|
|
|
self.mempool_fees = FeeHistogram()
|
|
self.fee_estimates = FeeTimeEstimates()
|
|
self.last_time_fee_estimates_requested = 0 # zero ensures immediate fees
|
|
|
|
def has_internet_connection(self) -> bool:
|
|
"""Our guess whether the device has Internet-connectivity."""
|
|
return self._has_ever_managed_to_connect_to_server
|
|
|
|
def has_channel_db(self):
|
|
return self.channel_db is not None
|
|
|
|
def start_gossip(self):
|
|
from . import lnrouter
|
|
from . import channel_db
|
|
from . import lnworker
|
|
if not self.config.LIGHTNING_USE_GOSSIP:
|
|
return
|
|
if self.lngossip is None:
|
|
self.channel_db = channel_db.ChannelDB(self)
|
|
self.path_finder = lnrouter.LNPathFinder(self.channel_db)
|
|
self.channel_db.load_data()
|
|
self.lngossip = lnworker.LNGossip(self.config)
|
|
self.lngossip.start_network(self)
|
|
|
|
async def stop_gossip(self, *, full_shutdown: bool = False):
|
|
if self.lngossip:
|
|
await self.lngossip.stop()
|
|
self.lngossip = None
|
|
self.channel_db.stop()
|
|
if full_shutdown:
|
|
await self.channel_db.stopped_event.wait()
|
|
self.channel_db = None
|
|
self.path_finder = None
|
|
|
|
@classmethod
|
|
def run_from_another_thread(cls, coro: 'Coroutine[Any, Any, T]', *, timeout=None) -> T:
|
|
loop = util.get_asyncio_loop()
|
|
assert util.get_running_loop() != loop, 'must not be called from asyncio thread'
|
|
fut = asyncio.run_coroutine_threadsafe(coro, loop)
|
|
return fut.result(timeout)
|
|
|
|
@staticmethod
|
|
def get_instance() -> Optional["Network"]:
|
|
"""Return the global singleton network instance.
|
|
Note that this can return None! If we are run with the --offline flag, there is no network.
|
|
"""
|
|
return _INSTANCE
|
|
|
|
def with_recent_servers_lock(func):
|
|
def func_wrapper(self, *args, **kwargs):
|
|
with self.recent_servers_lock:
|
|
return func(self, *args, **kwargs)
|
|
return func_wrapper
|
|
|
|
def _read_recent_servers(self) -> List[ServerAddr]:
|
|
if not self.config.path:
|
|
return []
|
|
path = os.path.join(self.config.path, "recent_servers")
|
|
try:
|
|
with open(path, "r", encoding='utf-8') as f:
|
|
data = f.read()
|
|
servers_list = json.loads(data)
|
|
return [ServerAddr.from_str(s) for s in servers_list]
|
|
except Exception:
|
|
return []
|
|
|
|
@with_recent_servers_lock
|
|
def _save_recent_servers(self):
|
|
if not self.config.path:
|
|
return
|
|
path = os.path.join(self.config.path, "recent_servers")
|
|
s = json.dumps(self._recent_servers, indent=4, sort_keys=True, default=str)
|
|
try:
|
|
with open(path, "w", encoding='utf-8') as f:
|
|
f.write(s)
|
|
except Exception:
|
|
pass
|
|
|
|
async def _server_is_lagging(self) -> bool:
|
|
sh = self.get_server_height()
|
|
if not sh:
|
|
self.logger.info('no height for main interface')
|
|
return True
|
|
lh = self.get_local_height()
|
|
result = (lh - sh) > 1
|
|
if result:
|
|
self.logger.info(f'{self.default_server} is lagging ({sh} vs {lh})')
|
|
return result
|
|
|
|
def _set_status(self, status):
|
|
self.connection_status = status
|
|
util.trigger_callback('status')
|
|
|
|
def is_connected(self):
|
|
interface = self.interface
|
|
return interface is not None and interface.is_connected_and_ready()
|
|
|
|
def is_connecting(self):
|
|
return self.connection_status == ConnectionState.CONNECTING
|
|
|
|
def get_connection_status_for_GUI(self):
|
|
ConnectionStates = {
|
|
ConnectionState.DISCONNECTED: _('Disconnected'),
|
|
ConnectionState.CONNECTING: _('Connecting'),
|
|
ConnectionState.CONNECTED: _('Connected'),
|
|
}
|
|
return ConnectionStates[self.connection_status]
|
|
|
|
async def _request_server_info(self, interface: 'Interface'):
|
|
await interface.ready
|
|
session = interface.session
|
|
|
|
async def get_banner():
|
|
self.banner = await interface.get_server_banner()
|
|
util.trigger_callback('banner', self.banner)
|
|
|
|
async def get_donation_address():
|
|
self.donation_address = await interface.get_donation_address()
|
|
|
|
async def get_server_peers():
|
|
server_peers = await session.send_request('server.peers.subscribe')
|
|
random.shuffle(server_peers)
|
|
max_accepted_peers = len(constants.net.DEFAULT_SERVERS) + NUM_RECENT_SERVERS
|
|
server_peers = server_peers[:max_accepted_peers]
|
|
# note that 'parse_servers' also validates the data (which is untrusted input!)
|
|
self.server_peers = parse_servers(server_peers)
|
|
util.trigger_callback('servers', self.get_servers())
|
|
|
|
async def get_relay_fee():
|
|
self.relay_fee = await interface.get_relay_fee()
|
|
|
|
async with OldTaskGroup() as group:
|
|
await group.spawn(get_banner)
|
|
await group.spawn(get_donation_address)
|
|
await group.spawn(get_server_peers)
|
|
await group.spawn(get_relay_fee)
|
|
await group.spawn(self._request_fee_estimates(interface))
|
|
|
|
async def _request_fee_estimates(self, interface):
|
|
self.requested_fee_estimates()
|
|
histogram = await interface.get_fee_histogram()
|
|
self.mempool_fees.set_data(histogram)
|
|
self.logger.info(f'fee_histogram {len(histogram)}')
|
|
util.trigger_callback('fee_histogram', self.mempool_fees)
|
|
|
|
def is_fee_estimates_update_required(self):
|
|
"""Checks time since last requested and updated fee estimates.
|
|
Returns True if an update should be requested.
|
|
"""
|
|
now = time.time()
|
|
return now - self.last_time_fee_estimates_requested > 60
|
|
|
|
def has_fee_etas(self):
|
|
return self.fee_estimates.has_data()
|
|
|
|
def has_fee_mempool(self) -> bool:
|
|
return self.mempool_fees.has_data()
|
|
|
|
def requested_fee_estimates(self):
|
|
self.last_time_fee_estimates_requested = time.time()
|
|
|
|
def get_parameters(self) -> NetworkParameters:
|
|
return NetworkParameters(server=self.default_server,
|
|
proxy=self.proxy,
|
|
auto_connect=self.auto_connect,
|
|
oneserver=self.oneserver)
|
|
|
|
def _init_parameters_from_config(self) -> None:
|
|
dns_hacks.configure_dns_resolver()
|
|
self.auto_connect = self.config.NETWORK_AUTO_CONNECT
|
|
if self.auto_connect and self.config.NETWORK_ONESERVER:
|
|
# enabling both oneserver and auto_connect doesn't really make sense
|
|
# assume oneserver is enabled for privacy reasons, disable auto_connect and assume server is unpredictable
|
|
self.logger.warning(f'both "oneserver" and "auto_connect" options enabled, disabling "auto_connect" and resetting "server".')
|
|
self.config.NETWORK_SERVER = "" # let _set_default_server set harmless default (localhost)
|
|
self.auto_connect = False
|
|
|
|
self._set_default_server()
|
|
self._set_proxy(ProxySettings.from_config(self.config))
|
|
self._maybe_set_oneserver()
|
|
|
|
def get_donation_address(self):
|
|
if self.is_connected():
|
|
return self.donation_address
|
|
|
|
def get_interfaces(self) -> List[ServerAddr]:
|
|
"""The list of servers for the connected interfaces."""
|
|
with self.interfaces_lock:
|
|
return list(self.interfaces)
|
|
|
|
def get_status(self):
|
|
n = len(self.get_interfaces())
|
|
return _("Connected to {0} nodes.").format(n) if n > 1 else _("Connected to {0} node.").format(n) if n == 1 else _("Not connected")
|
|
|
|
def get_fee_estimates(self):
|
|
from statistics import median
|
|
if self.auto_connect:
|
|
with self.interfaces_lock:
|
|
out = {}
|
|
for n in FEE_ETA_TARGETS[0:-1]:
|
|
try:
|
|
out[n] = int(median(filter(None, [i.fee_estimates_eta.get(n) for i in self.interfaces.values()])))
|
|
except Exception:
|
|
continue
|
|
return out
|
|
else:
|
|
if not self.interface:
|
|
return {}
|
|
return self.interface.fee_estimates_eta
|
|
|
|
def update_fee_estimates(self, *, fee_est: Dict[int, int] = None):
|
|
if fee_est is None:
|
|
if self.config.TEST_DISABLE_AUTOMATIC_FEE_ETA_UPDATE:
|
|
return
|
|
fee_est = self.get_fee_estimates()
|
|
for nblock_target, fee in fee_est.items():
|
|
self.fee_estimates.set_data(nblock_target, fee)
|
|
if not hasattr(self, "_prev_fee_est") or self._prev_fee_est != fee_est:
|
|
self._prev_fee_est = copy.copy(fee_est)
|
|
self.logger.info(f'fee_estimates {fee_est}')
|
|
util.trigger_callback('fee', self.fee_estimates)
|
|
|
|
|
|
@with_recent_servers_lock
|
|
def get_servers(self):
|
|
# note: order of sources when adding servers here is crucial!
|
|
# don't let "server_peers" overwrite anything,
|
|
# otherwise main server can eclipse the client
|
|
out = dict()
|
|
# add servers received from main interface
|
|
server_peers = self.server_peers
|
|
if server_peers:
|
|
out.update(filter_version(server_peers.copy()))
|
|
# hardcoded servers
|
|
out.update(constants.net.DEFAULT_SERVERS)
|
|
# add recent servers
|
|
for server in self._recent_servers:
|
|
port = str(server.port)
|
|
if server.host in out:
|
|
out[server.host].update({server.protocol: port})
|
|
else:
|
|
out[server.host] = {server.protocol: port}
|
|
# add bookmarks
|
|
bookmarks = self.config.NETWORK_BOOKMARKED_SERVERS or []
|
|
for server_str in bookmarks:
|
|
try:
|
|
server = ServerAddr.from_str(server_str)
|
|
except ValueError:
|
|
continue
|
|
port = str(server.port)
|
|
if server.host in out:
|
|
out[server.host].update({server.protocol: port})
|
|
else:
|
|
out[server.host] = {server.protocol: port}
|
|
# potentially filter out some
|
|
if self.config.NETWORK_NOONION:
|
|
out = filter_noonion(out)
|
|
return out
|
|
|
|
def _get_next_server_to_try(self) -> Optional[ServerAddr]:
|
|
now = time.time()
|
|
with self.interfaces_lock:
|
|
connected_servers = set(self.interfaces) | self._connecting_ifaces | self._closing_ifaces
|
|
# First try from recent servers. (which are persisted)
|
|
# As these are servers we successfully connected to recently, they are
|
|
# most likely to work. This also makes servers "sticky".
|
|
# Note: with sticky servers, it is more difficult for an attacker to eclipse the client,
|
|
# however if they succeed, the eclipsing would persist. To try to balance this,
|
|
# we only give priority to recent_servers up to NUM_STICKY_SERVERS.
|
|
with self.recent_servers_lock:
|
|
recent_servers = list(self._recent_servers)
|
|
recent_servers = [s for s in recent_servers if s.protocol in self._allowed_protocols]
|
|
if len(connected_servers & set(recent_servers)) < NUM_STICKY_SERVERS:
|
|
for server in recent_servers:
|
|
if server in connected_servers:
|
|
continue
|
|
if not self._can_retry_addr(server, now=now):
|
|
continue
|
|
return server
|
|
# try all servers we know about, pick one at random
|
|
hostmap = self.get_servers()
|
|
servers = list(set(filter_protocol(hostmap, allowed_protocols=self._allowed_protocols)) - connected_servers)
|
|
random.shuffle(servers)
|
|
for server in servers:
|
|
if not self._can_retry_addr(server, now=now):
|
|
continue
|
|
return server
|
|
return None
|
|
|
|
def _set_default_server(self) -> None:
|
|
# Server for addresses and transactions
|
|
server = self.config.NETWORK_SERVER
|
|
# Sanitize default server
|
|
if server:
|
|
try:
|
|
self.default_server = ServerAddr.from_str(server)
|
|
except Exception:
|
|
self.logger.warning(f'failed to parse server-string ({server!r}); falling back to localhost:1:s.')
|
|
self.default_server = ServerAddr.from_str("localhost:1:s")
|
|
else:
|
|
# if oneserver is enabled but no server specified then don't pick a random server
|
|
if self.config.NETWORK_ONESERVER:
|
|
self.logger.warning(f'"oneserver" option enabled, but no "server" defined; falling back to localhost:1:s.')
|
|
self.default_server = ServerAddr.from_str("localhost:1:s")
|
|
else:
|
|
self.default_server = pick_random_server(allowed_protocols=self._allowed_protocols)
|
|
assert isinstance(self.default_server, ServerAddr), f"invalid type for default_server: {self.default_server!r}"
|
|
|
|
def _set_proxy(self, proxy: ProxySettings):
|
|
if self.proxy == proxy:
|
|
return
|
|
|
|
self.logger.info(f'setting proxy {proxy}')
|
|
self.proxy = proxy
|
|
|
|
# reset is_proxy_tor to unknown, and re-detect it:
|
|
self.is_proxy_tor = None
|
|
self._detect_if_proxy_is_tor()
|
|
|
|
util.trigger_callback('proxy_set', self.proxy)
|
|
|
|
def _detect_if_proxy_is_tor(self) -> None:
|
|
async def tor_probe_task(p):
|
|
assert p is not None
|
|
is_tor = await util.is_tor_socks_port(p.host, int(p.port))
|
|
if self.proxy == p: # is this the proxy we probed?
|
|
if self.is_proxy_tor != is_tor:
|
|
self.logger.info(f'Proxy is {"" if is_tor else "not "}TOR')
|
|
self.is_proxy_tor = is_tor
|
|
util.trigger_callback('tor_probed', is_tor)
|
|
|
|
proxy = self.proxy
|
|
if proxy and proxy.enabled and proxy.mode == 'socks5':
|
|
asyncio.run_coroutine_threadsafe(tor_probe_task(proxy), self.asyncio_loop)
|
|
|
|
@log_exceptions
|
|
async def set_parameters(self, net_params: NetworkParameters):
|
|
proxy = net_params.proxy
|
|
proxy_str = proxy.serialize_proxy_cfgstr()
|
|
proxy_enabled = proxy.enabled
|
|
proxy_user = proxy.user
|
|
proxy_pass = proxy.password
|
|
server = net_params.server
|
|
# sanitize parameters
|
|
try:
|
|
if proxy:
|
|
# proxy_modes.index(proxy['mode']) + 1
|
|
ProxySettings.MODES.index(proxy.mode) + 1
|
|
# int(proxy['port'])
|
|
int(proxy.port)
|
|
except Exception:
|
|
proxy.enabled = False
|
|
# return
|
|
self.config.NETWORK_AUTO_CONNECT = net_params.auto_connect
|
|
self.config.NETWORK_ONESERVER = net_params.oneserver
|
|
self.config.NETWORK_PROXY_ENABLED = proxy_enabled
|
|
self.config.NETWORK_PROXY = proxy_str
|
|
self.config.NETWORK_PROXY_USER = proxy_user
|
|
self.config.NETWORK_PROXY_PASSWORD = proxy_pass
|
|
self.config.NETWORK_SERVER = str(server)
|
|
# abort if changes were not allowed by config
|
|
if self.config.NETWORK_SERVER != str(server) \
|
|
or self.config.NETWORK_PROXY_ENABLED != proxy_enabled \
|
|
or self.config.NETWORK_PROXY != proxy_str \
|
|
or self.config.NETWORK_PROXY_USER != proxy_user \
|
|
or self.config.NETWORK_PROXY_PASSWORD != proxy_pass \
|
|
or self.config.NETWORK_ONESERVER != net_params.oneserver:
|
|
return
|
|
|
|
proxy_changed = self.proxy != proxy
|
|
oneserver_changed = self.oneserver != net_params.oneserver
|
|
default_server_changed = self.default_server != server
|
|
self._init_parameters_from_config()
|
|
if not self._was_started:
|
|
return
|
|
|
|
async with self.restart_lock:
|
|
if proxy_changed or oneserver_changed:
|
|
# Restart the network
|
|
await self.stop(full_shutdown=False)
|
|
await self._start()
|
|
elif default_server_changed:
|
|
await self.switch_to_interface(server)
|
|
else:
|
|
await self.switch_lagging_interface()
|
|
util.trigger_callback('network_updated')
|
|
|
|
def _maybe_set_oneserver(self) -> None:
|
|
oneserver = self.config.NETWORK_ONESERVER
|
|
self.oneserver = oneserver
|
|
self.num_server = NUM_TARGET_CONNECTED_SERVERS if not oneserver else 0
|
|
|
|
def is_server_bookmarked(self, server: ServerAddr) -> bool:
|
|
bookmarks = self.config.NETWORK_BOOKMARKED_SERVERS or []
|
|
return str(server) in bookmarks
|
|
|
|
def set_server_bookmark(self, server: ServerAddr, *, add: bool) -> None:
|
|
server_str = str(server)
|
|
with self.config.lock:
|
|
bookmarks = self.config.NETWORK_BOOKMARKED_SERVERS or []
|
|
if add:
|
|
if server_str not in bookmarks:
|
|
bookmarks.append(server_str)
|
|
else: # remove
|
|
if server_str in bookmarks:
|
|
bookmarks.remove(server_str)
|
|
self.config.NETWORK_BOOKMARKED_SERVERS = bookmarks
|
|
|
|
async def _switch_to_random_interface(self):
|
|
'''Switch to a random connected server other than the current one'''
|
|
servers = self.get_interfaces() # Those in connected state
|
|
if self.default_server in servers:
|
|
servers.remove(self.default_server)
|
|
if servers:
|
|
await self.switch_to_interface(random.choice(servers))
|
|
|
|
async def switch_lagging_interface(self):
|
|
"""If auto_connect and lagging, switch interface (only within fork)."""
|
|
if self.auto_connect and await self._server_is_lagging():
|
|
# switch to one that has the correct header (not height)
|
|
best_header = self.blockchain().header_at_tip()
|
|
with self.interfaces_lock: interfaces = list(self.interfaces.values())
|
|
filtered = list(filter(lambda iface: iface.tip_header == best_header, interfaces))
|
|
if filtered:
|
|
chosen_iface = random.choice(filtered)
|
|
await self.switch_to_interface(chosen_iface.server)
|
|
|
|
async def switch_unwanted_fork_interface(self) -> None:
|
|
"""If auto_connect, maybe switch to another fork/chain."""
|
|
if not self.auto_connect or not self.interface:
|
|
return
|
|
with self.interfaces_lock: interfaces = list(self.interfaces.values())
|
|
pref_height = self._blockchain_preferred_block['height']
|
|
pref_hash = self._blockchain_preferred_block['hash']
|
|
# shortcut for common case
|
|
if pref_height == 0:
|
|
return
|
|
# maybe try switching chains; starting with most desirable first
|
|
matching_chains = blockchain.get_chains_that_contain_header(pref_height, pref_hash)
|
|
chains_to_try = list(matching_chains) + [blockchain.get_best_chain()]
|
|
for rank, chain in enumerate(chains_to_try):
|
|
# check if main interface is already on this fork
|
|
if self.interface.blockchain == chain:
|
|
return
|
|
# switch to another random interface that is on this fork, if any
|
|
filtered = [iface for iface in interfaces
|
|
if iface.blockchain == chain]
|
|
if filtered:
|
|
self.logger.info(f"switching to (more) preferred fork (rank {rank})")
|
|
chosen_iface = random.choice(filtered)
|
|
await self.switch_to_interface(chosen_iface.server)
|
|
return
|
|
self.logger.info("tried to switch to (more) preferred fork but no interfaces are on any")
|
|
|
|
async def switch_to_interface(self, server: ServerAddr):
|
|
"""Switch to server as our main interface. If no connection exists,
|
|
queue interface to be started. The actual switch will
|
|
happen when the interface becomes ready.
|
|
"""
|
|
assert isinstance(server, ServerAddr), f"expected ServerAddr, got {type(server)}"
|
|
self.default_server = server
|
|
old_interface = self.interface
|
|
old_server = old_interface.server if old_interface else None
|
|
|
|
# Stop any current interface in order to terminate subscriptions,
|
|
# and to cancel tasks in interface.taskgroup.
|
|
if old_server and old_server != server:
|
|
# don't wait for old_interface to close as that might be slow:
|
|
await self.taskgroup.spawn(self._close_interface(old_interface))
|
|
|
|
if server not in self.interfaces:
|
|
self.interface = None
|
|
await self.taskgroup.spawn(self._run_new_interface(server))
|
|
return
|
|
|
|
i = self.interfaces[server]
|
|
if old_interface != i:
|
|
if not i.is_connected_and_ready():
|
|
return
|
|
self.logger.info(f"switching to {server}")
|
|
blockchain_updated = i.blockchain != self.blockchain()
|
|
self.interface = i
|
|
try:
|
|
await i.taskgroup.spawn(self._request_server_info(i))
|
|
except RuntimeError as e: # see #7677
|
|
if len(e.args) >= 1 and e.args[0] == 'task group terminated':
|
|
self.logger.warning(f"tried to switch to {server} but interface.taskgroup is already dead.")
|
|
self.interface = None
|
|
return
|
|
raise
|
|
util.trigger_callback('default_server_changed')
|
|
self.default_server_changed_event.set()
|
|
self.default_server_changed_event.clear()
|
|
self._set_status(ConnectionState.CONNECTED)
|
|
util.trigger_callback('network_updated')
|
|
if blockchain_updated:
|
|
util.trigger_callback('blockchain_updated')
|
|
|
|
async def _close_interface(self, interface: Optional[Interface]):
|
|
if not interface:
|
|
return
|
|
if interface.server in self._closing_ifaces:
|
|
return
|
|
self._closing_ifaces.add(interface.server)
|
|
with self.interfaces_lock:
|
|
if self.interfaces.get(interface.server) == interface:
|
|
self.interfaces.pop(interface.server)
|
|
if interface == self.interface:
|
|
self.interface = None
|
|
try:
|
|
# this can take some time if server/connection is slow:
|
|
await interface.close()
|
|
await interface.got_disconnected.wait()
|
|
finally:
|
|
self._closing_ifaces.discard(interface.server)
|
|
|
|
@with_recent_servers_lock
|
|
def _add_recent_server(self, server: ServerAddr) -> None:
|
|
self._on_connection_successfully_established(server)
|
|
# list is ordered
|
|
if server in self._recent_servers:
|
|
self._recent_servers.remove(server)
|
|
self._recent_servers.insert(0, server)
|
|
self._recent_servers = self._recent_servers[:NUM_RECENT_SERVERS]
|
|
self._save_recent_servers()
|
|
|
|
async def connection_down(self, interface: Interface):
|
|
'''A connection to server either went down, or was never made.
|
|
We distinguish by whether it is in self.interfaces.'''
|
|
if not interface: return
|
|
if interface.server == self.default_server:
|
|
self._set_status(ConnectionState.DISCONNECTED)
|
|
await self._close_interface(interface)
|
|
util.trigger_callback('network_updated')
|
|
|
|
def get_network_timeout_seconds(self, request_type=NetworkTimeout.Generic) -> int:
|
|
if self.config.NETWORK_TIMEOUT:
|
|
return self.config.NETWORK_TIMEOUT
|
|
if self.oneserver and not self.auto_connect:
|
|
return request_type.MOST_RELAXED
|
|
if self.proxy and self.proxy.enabled:
|
|
return request_type.RELAXED
|
|
return request_type.NORMAL
|
|
|
|
@ignore_exceptions # do not kill outer taskgroup
|
|
@log_exceptions
|
|
async def _run_new_interface(self, server: ServerAddr):
|
|
assert isinstance(server, ServerAddr), f"expected ServerAddr, got {type(server)}"
|
|
if (server in self.interfaces
|
|
or server in self._connecting_ifaces
|
|
or server in self._closing_ifaces):
|
|
return
|
|
self._connecting_ifaces.add(server)
|
|
if server == self.default_server:
|
|
self.logger.info(f"connecting to {server} as new interface")
|
|
self._set_status(ConnectionState.CONNECTING)
|
|
self._trying_addr_now(server)
|
|
|
|
interface = Interface(network=self, server=server)
|
|
# note: using longer timeouts here as DNS can sometimes be slow!
|
|
timeout = self.get_network_timeout_seconds(NetworkTimeout.Generic)
|
|
try:
|
|
await util.wait_for2(interface.ready, timeout)
|
|
except BaseException as e:
|
|
self.logger.info(f"couldn't launch iface {server} -- {repr(e)}")
|
|
await interface.close()
|
|
return
|
|
else:
|
|
with self.interfaces_lock:
|
|
assert server not in self.interfaces
|
|
self.interfaces[server] = interface
|
|
finally:
|
|
self._connecting_ifaces.discard(server)
|
|
|
|
if server == self.default_server:
|
|
await self.switch_to_interface(server)
|
|
|
|
self._has_ever_managed_to_connect_to_server = True
|
|
self._add_recent_server(server)
|
|
util.trigger_callback('network_updated')
|
|
# When the proxy settings were set, the proxy (if any) might have been unreachable,
|
|
# resulting in a false-negative for Tor-detection. Given we just connected to a server, re-test now.
|
|
self._detect_if_proxy_is_tor()
|
|
|
|
def check_interface_against_healthy_spread_of_connected_servers(self, iface_to_check: Interface) -> bool:
|
|
# main interface is exempt. this makes switching servers easier
|
|
if iface_to_check.is_main_server():
|
|
return True
|
|
if not iface_to_check.bucket_based_on_ipaddress():
|
|
return True
|
|
# bucket connected interfaces
|
|
with self.interfaces_lock:
|
|
interfaces = list(self.interfaces.values())
|
|
if iface_to_check in interfaces:
|
|
interfaces.remove(iface_to_check)
|
|
buckets = defaultdict(list)
|
|
for iface in interfaces:
|
|
buckets[iface.bucket_based_on_ipaddress()].append(iface)
|
|
# check proposed server against buckets
|
|
onion_servers = buckets[BUCKET_NAME_OF_ONION_SERVERS]
|
|
if iface_to_check.is_tor():
|
|
# keep number of onion servers below half of all connected servers
|
|
if len(onion_servers) > NUM_TARGET_CONNECTED_SERVERS // 2:
|
|
return False
|
|
else:
|
|
bucket = iface_to_check.bucket_based_on_ipaddress()
|
|
if len(buckets[bucket]) > 0:
|
|
return False
|
|
return True
|
|
|
|
def best_effort_reliable(func):
|
|
@functools.wraps(func)
|
|
async def make_reliable_wrapper(self: 'Network', *args, **kwargs):
|
|
for i in range(10):
|
|
iface = self.interface
|
|
# retry until there is a main interface
|
|
if not iface:
|
|
async with ignore_after(1):
|
|
await self.default_server_changed_event.wait()
|
|
continue # try again
|
|
assert iface.ready.done(), "interface not ready yet"
|
|
# try actual request
|
|
try:
|
|
async with OldTaskGroup(wait=any) as group:
|
|
task = await group.spawn(func(self, *args, **kwargs))
|
|
await group.spawn(iface.got_disconnected.wait())
|
|
except RequestTimedOut:
|
|
await iface.close()
|
|
await iface.got_disconnected.wait()
|
|
continue # try again
|
|
except RequestCorrupted as e:
|
|
# TODO ban server?
|
|
iface.logger.exception(f"RequestCorrupted: {e}")
|
|
await iface.close()
|
|
await iface.got_disconnected.wait()
|
|
continue # try again
|
|
if task.done() and not task.cancelled():
|
|
return task.result()
|
|
# otherwise; try again
|
|
raise BestEffortRequestFailed('cannot establish a connection... gave up.')
|
|
return make_reliable_wrapper
|
|
|
|
def catch_server_exceptions(func):
|
|
"""Decorator that wraps server errors in UntrustedServerReturnedError,
|
|
to avoid showing untrusted arbitrary text to users.
|
|
"""
|
|
@functools.wraps(func)
|
|
async def wrapper(self, *args, **kwargs):
|
|
try:
|
|
return await func(self, *args, **kwargs)
|
|
except aiorpcx.jsonrpc.CodeMessageError as e:
|
|
wrapped_exc = UntrustedServerReturnedError(original_exception=e)
|
|
# log (sanitized) untrusted error text now, to ease debugging
|
|
self.logger.debug(f"got error from server for {func.__qualname__}: {wrapped_exc.get_untrusted_message()!r}")
|
|
raise wrapped_exc from e
|
|
return wrapper
|
|
|
|
@best_effort_reliable
|
|
@catch_server_exceptions
|
|
async def get_merkle_for_transaction(self, tx_hash: str, tx_height: int) -> dict:
|
|
if self.interface is None: # handled by best_effort_reliable
|
|
raise RequestTimedOut()
|
|
return await self.interface.get_merkle_for_transaction(tx_hash=tx_hash, tx_height=tx_height)
|
|
|
|
@best_effort_reliable
|
|
async def broadcast_transaction(self, tx: 'Transaction', *, timeout=None) -> None:
|
|
"""caller should handle TxBroadcastError"""
|
|
if self.interface is None: # handled by best_effort_reliable
|
|
raise RequestTimedOut()
|
|
await self.interface.broadcast_transaction(tx, timeout=timeout)
|
|
|
|
async def try_broadcasting(self, tx: 'Transaction', name: str) -> bool:
|
|
try:
|
|
await self.broadcast_transaction(tx)
|
|
except Exception as e:
|
|
self.logger.info(f'error: could not broadcast {name} {tx.txid()}, {str(e)}')
|
|
return False
|
|
else:
|
|
self.logger.info(f'success: broadcasting {name} {tx.txid()}')
|
|
return True
|
|
|
|
@best_effort_reliable
|
|
@catch_server_exceptions
|
|
async def get_transaction(self, tx_hash: str, *, timeout=None) -> str:
|
|
if self.interface is None: # handled by best_effort_reliable
|
|
raise RequestTimedOut()
|
|
return await self.interface.get_transaction(tx_hash=tx_hash, timeout=timeout)
|
|
|
|
@best_effort_reliable
|
|
@catch_server_exceptions
|
|
async def get_history_for_scripthash(self, sh: str) -> List[dict]:
|
|
if self.interface is None: # handled by best_effort_reliable
|
|
raise RequestTimedOut()
|
|
return await self.interface.get_history_for_scripthash(sh)
|
|
|
|
@best_effort_reliable
|
|
@catch_server_exceptions
|
|
async def listunspent_for_scripthash(self, sh: str) -> List[dict]:
|
|
if self.interface is None: # handled by best_effort_reliable
|
|
raise RequestTimedOut()
|
|
return await self.interface.listunspent_for_scripthash(sh)
|
|
|
|
@best_effort_reliable
|
|
@catch_server_exceptions
|
|
async def get_balance_for_scripthash(self, sh: str) -> dict:
|
|
if self.interface is None: # handled by best_effort_reliable
|
|
raise RequestTimedOut()
|
|
return await self.interface.get_balance_for_scripthash(sh)
|
|
|
|
@best_effort_reliable
|
|
@catch_server_exceptions
|
|
async def get_txid_from_txpos(self, tx_height, tx_pos, merkle):
|
|
if self.interface is None: # handled by best_effort_reliable
|
|
raise RequestTimedOut()
|
|
return await self.interface.get_txid_from_txpos(tx_height, tx_pos, merkle)
|
|
|
|
def blockchain(self) -> Blockchain:
|
|
interface = self.interface
|
|
if interface and interface.blockchain is not None:
|
|
self._blockchain = interface.blockchain
|
|
return self._blockchain
|
|
|
|
def get_blockchains(self):
|
|
out = {} # blockchain_id -> list(interfaces)
|
|
with blockchain.blockchains_lock: blockchain_items = list(blockchain.blockchains.items())
|
|
with self.interfaces_lock: interfaces_values = list(self.interfaces.values())
|
|
for chain_id, bc in blockchain_items:
|
|
r = list(filter(lambda i: i.blockchain==bc, interfaces_values))
|
|
if r:
|
|
out[chain_id] = r
|
|
return out
|
|
|
|
def _set_preferred_chain(self, chain: Optional[Blockchain]):
|
|
if chain:
|
|
height = chain.get_max_forkpoint()
|
|
header_hash = chain.get_hash(height)
|
|
else:
|
|
height = 0
|
|
header_hash = constants.net.GENESIS
|
|
self._blockchain_preferred_block = {
|
|
'height': height,
|
|
'hash': header_hash,
|
|
}
|
|
self.config.BLOCKCHAIN_PREFERRED_BLOCK = self._blockchain_preferred_block
|
|
|
|
async def follow_chain_given_id(self, chain_id: str) -> None:
|
|
bc = blockchain.blockchains.get(chain_id)
|
|
if not bc:
|
|
raise Exception('blockchain {} not found'.format(chain_id))
|
|
self._set_preferred_chain(bc)
|
|
# select server on this chain
|
|
with self.interfaces_lock: interfaces = list(self.interfaces.values())
|
|
interfaces_on_selected_chain = list(filter(lambda iface: iface.blockchain == bc, interfaces))
|
|
if len(interfaces_on_selected_chain) == 0: return
|
|
chosen_iface = random.choice(interfaces_on_selected_chain) # type: Interface
|
|
# switch to server (and save to config)
|
|
net_params = self.get_parameters()
|
|
# we select a random interface, so set connection mode back to autoconnect
|
|
net_params = net_params._replace(server=chosen_iface.server, auto_connect=True, oneserver=False)
|
|
await self.set_parameters(net_params)
|
|
|
|
def follow_chain_given_server(self, server: ServerAddr) -> None:
|
|
# note that server_str should correspond to a connected interface
|
|
iface = self.interfaces[server]
|
|
self._set_preferred_chain(iface.blockchain)
|
|
self.logger.debug(f"following {self.config.BLOCKCHAIN_PREFERRED_BLOCK=}")
|
|
|
|
def get_server_height(self) -> int:
|
|
"""Length of header chain, as claimed by main interface."""
|
|
interface = self.interface
|
|
return interface.tip if interface else 0
|
|
|
|
def get_local_height(self) -> int:
|
|
"""Length of header chain, POW-verified.
|
|
In case of a chain split, this is for the branch the main interface is on,
|
|
but it is the tip of that branch (even if main interface is behind).
|
|
"""
|
|
return self.blockchain().height()
|
|
|
|
def export_checkpoints(self, path):
|
|
"""Run manually to generate blockchain checkpoints.
|
|
Kept for console use only.
|
|
"""
|
|
cp = self.blockchain().get_checkpoints()
|
|
with open(path, 'w', encoding='utf-8') as f:
|
|
f.write(json.dumps(cp, indent=4))
|
|
|
|
async def _start(self):
|
|
assert not self.taskgroup
|
|
self.taskgroup = taskgroup = OldTaskGroup()
|
|
assert not self.interface and not self.interfaces
|
|
assert not self._connecting_ifaces
|
|
assert not self._closing_ifaces
|
|
self.logger.info('starting network')
|
|
self._clear_addr_retry_times()
|
|
self._init_parameters_from_config()
|
|
await self.taskgroup.spawn(self._run_new_interface(self.default_server))
|
|
|
|
async def main():
|
|
self.logger.info(f"starting taskgroup ({hex(id(taskgroup))}).")
|
|
try:
|
|
# note: if a task finishes with CancelledError, that
|
|
# will NOT raise, and the group will keep the other tasks running
|
|
async with taskgroup as group:
|
|
await group.spawn(self._maintain_sessions())
|
|
[await group.spawn(job) for job in self._jobs]
|
|
except Exception as e:
|
|
self.logger.exception(f"taskgroup died ({hex(id(taskgroup))}).")
|
|
finally:
|
|
self.logger.info(f"taskgroup stopped ({hex(id(taskgroup))}).")
|
|
asyncio.run_coroutine_threadsafe(main(), self.asyncio_loop)
|
|
|
|
util.trigger_callback('network_updated')
|
|
|
|
def start(self, jobs: Iterable = None):
|
|
"""Schedule starting the network, along with the given job co-routines.
|
|
|
|
Note: the jobs will *restart* every time the network restarts, e.g. on proxy
|
|
setting changes.
|
|
"""
|
|
self._was_started = True
|
|
self._jobs = jobs or []
|
|
asyncio.run_coroutine_threadsafe(self._start(), self.asyncio_loop)
|
|
|
|
@log_exceptions
|
|
async def stop(self, *, full_shutdown: bool = True):
|
|
if not self._was_started:
|
|
self.logger.info("not stopping network as it was never started")
|
|
return
|
|
self.logger.info("stopping network")
|
|
# timeout: if full_shutdown, it is up to the caller to time us out,
|
|
# otherwise if e.g. restarting due to proxy changes, we time out fast
|
|
async with (nullcontext() if full_shutdown else ignore_after(1)):
|
|
async with OldTaskGroup() as group:
|
|
await group.spawn(self.taskgroup.cancel_remaining())
|
|
if full_shutdown:
|
|
await group.spawn(self.stop_gossip(full_shutdown=full_shutdown))
|
|
self.taskgroup = None
|
|
self.interface = None
|
|
self.interfaces = {}
|
|
self._connecting_ifaces.clear()
|
|
self._closing_ifaces.clear()
|
|
if not full_shutdown:
|
|
util.trigger_callback('network_updated')
|
|
|
|
async def _ensure_there_is_a_main_interface(self):
|
|
if self.interface:
|
|
return
|
|
# if auto_connect is set, try a different server
|
|
if self.auto_connect and not self.is_connecting():
|
|
await self._switch_to_random_interface()
|
|
# if auto_connect is not set, or still no main interface, retry current
|
|
if not self.interface and not self.is_connecting():
|
|
if self._can_retry_addr(self.default_server, urgent=True):
|
|
await self.switch_to_interface(self.default_server)
|
|
|
|
async def _maintain_sessions(self):
|
|
async def maybe_start_new_interfaces():
|
|
num_existing_ifaces = len(self.interfaces) + len(self._connecting_ifaces) + len(self._closing_ifaces)
|
|
for i in range(self.num_server - num_existing_ifaces):
|
|
# FIXME this should try to honour "healthy spread of connected servers"
|
|
server = self._get_next_server_to_try()
|
|
if server:
|
|
assert isinstance(server, ServerAddr), f"expected ServerAddr, got {type(server)}"
|
|
await self.taskgroup.spawn(self._run_new_interface(server))
|
|
async def maintain_healthy_spread_of_connected_servers():
|
|
with self.interfaces_lock: interfaces = list(self.interfaces.values())
|
|
random.shuffle(interfaces)
|
|
for iface in interfaces:
|
|
if not self.check_interface_against_healthy_spread_of_connected_servers(iface):
|
|
self.logger.info(f"disconnecting from {iface.server}. too many connected "
|
|
f"servers already in bucket {iface.bucket_based_on_ipaddress()}")
|
|
await self._close_interface(iface)
|
|
async def maintain_main_interface():
|
|
await self._ensure_there_is_a_main_interface()
|
|
if self.is_connected():
|
|
if self.is_fee_estimates_update_required():
|
|
await self.interface.taskgroup.spawn(self._request_fee_estimates, self.interface)
|
|
|
|
while True:
|
|
await maybe_start_new_interfaces()
|
|
await maintain_healthy_spread_of_connected_servers()
|
|
await maintain_main_interface()
|
|
await asyncio.sleep(0.1)
|
|
|
|
@classmethod
|
|
async def async_send_http_on_proxy(
|
|
cls, method: str, url: str, *,
|
|
params: dict = None,
|
|
body: bytes = None,
|
|
json: dict = None,
|
|
headers=None,
|
|
on_finish=None,
|
|
timeout=None,
|
|
):
|
|
async def default_on_finish(resp: ClientResponse):
|
|
resp.raise_for_status()
|
|
return await resp.text()
|
|
if headers is None:
|
|
headers = {}
|
|
if on_finish is None:
|
|
on_finish = default_on_finish
|
|
network = cls.get_instance()
|
|
proxy = network.proxy if network else None
|
|
async with make_aiohttp_session(proxy, timeout=timeout) as session:
|
|
if method == 'get':
|
|
async with session.get(url, params=params, headers=headers) as resp:
|
|
return await on_finish(resp)
|
|
elif method == 'post':
|
|
assert body is not None or json is not None, 'body or json must be supplied if method is post'
|
|
if body is not None:
|
|
async with session.post(url, data=body, headers=headers) as resp:
|
|
return await on_finish(resp)
|
|
elif json is not None:
|
|
async with session.post(url, json=json, headers=headers) as resp:
|
|
return await on_finish(resp)
|
|
else:
|
|
raise Exception(f"unexpected {method=!r}")
|
|
|
|
@classmethod
|
|
def send_http_on_proxy(cls, method, url, **kwargs):
|
|
loop = util.get_asyncio_loop()
|
|
assert util.get_running_loop() != loop, 'must not be called from asyncio thread'
|
|
coro = asyncio.run_coroutine_threadsafe(cls.async_send_http_on_proxy(method, url, **kwargs), loop)
|
|
# note: _send_http_on_proxy has its own timeout, so no timeout here:
|
|
return coro.result()
|
|
|
|
# methods used in scripts
|
|
async def get_peers(self):
|
|
while not self.is_connected():
|
|
await asyncio.sleep(1)
|
|
session = self.interface.session
|
|
return parse_servers(await session.send_request('server.peers.subscribe'))
|
|
|
|
async def send_multiple_requests(
|
|
self,
|
|
servers: Sequence[ServerAddr],
|
|
method: str,
|
|
params: Sequence,
|
|
*,
|
|
timeout: int = None,
|
|
):
|
|
if timeout is None:
|
|
timeout = self.get_network_timeout_seconds(NetworkTimeout.Urgent)
|
|
responses = dict()
|
|
async def get_response(server: ServerAddr):
|
|
interface = Interface(network=self, server=server)
|
|
try:
|
|
await util.wait_for2(interface.ready, timeout)
|
|
except BaseException as e:
|
|
await interface.close()
|
|
return
|
|
try:
|
|
res = await interface.session.send_request(method, params, timeout=10)
|
|
except Exception as e:
|
|
res = e
|
|
responses[interface.server] = res
|
|
async with OldTaskGroup() as group:
|
|
for server in servers:
|
|
await group.spawn(get_response(server))
|
|
return responses
|
|
|
|
async def prune_offline_servers(self, hostmap):
|
|
peers = filter_protocol(hostmap, allowed_protocols=("t", "s",))
|
|
timeout = self.get_network_timeout_seconds(NetworkTimeout.Generic)
|
|
replies = await self.send_multiple_requests(peers, 'blockchain.headers.subscribe', [], timeout=timeout)
|
|
servers_replied = {serveraddr.host for serveraddr in replies.keys()}
|
|
servers_dict = {k: v for k, v in hostmap.items()
|
|
if k in servers_replied}
|
|
return servers_dict
|