forked from LBRYCommunity/lbry-sdk
507 lines
19 KiB
Python
507 lines
19 KiB
Python
# Copyright (c) 2017-2018, Neil Booth
|
|
#
|
|
# All rights reserved.
|
|
#
|
|
# See the file "LICENCE" for information about the copyright
|
|
# and warranty status of this software.
|
|
|
|
"""Peer management."""
|
|
|
|
import asyncio
|
|
import random
|
|
import socket
|
|
import ssl
|
|
import time
|
|
import typing
|
|
from asyncio import Event, sleep
|
|
from collections import defaultdict, Counter
|
|
|
|
from lbry.wallet.tasks import TaskGroup
|
|
from lbry.wallet.rpc import (
|
|
Connector, RPCSession, SOCKSProxy, Notification, handler_invocation,
|
|
SOCKSError, RPCError
|
|
)
|
|
from lbry.wallet.server.peer import Peer
|
|
from lbry.wallet.server.util import class_logger, protocol_tuple
|
|
|
|
PEER_GOOD, PEER_STALE, PEER_NEVER, PEER_BAD = range(4)
|
|
STALE_SECS = 24 * 3600
|
|
WAKEUP_SECS = 300
|
|
|
|
|
|
class BadPeerError(Exception):
|
|
pass
|
|
|
|
|
|
def assert_good(message, result, instance):
|
|
if not isinstance(result, instance):
|
|
raise BadPeerError(f'{message} returned bad result type '
|
|
f'{type(result).__name__}')
|
|
|
|
|
|
class PeerSession(RPCSession):
|
|
"""An outgoing session to a peer."""
|
|
|
|
async def handle_request(self, request):
|
|
# We subscribe so might be unlucky enough to get a notification...
|
|
if (isinstance(request, Notification) and
|
|
request.method == 'blockchain.headers.subscribe'):
|
|
pass
|
|
else:
|
|
await handler_invocation(None, request) # Raises
|
|
|
|
|
|
class PeerManager:
|
|
"""Looks after the DB of peer network servers.
|
|
|
|
Attempts to maintain a connection with up to 8 peers.
|
|
Issues a 'peers.subscribe' RPC to them and tells them our data.
|
|
"""
|
|
def __init__(self, env, db):
|
|
self.logger = class_logger(__name__, self.__class__.__name__)
|
|
# Initialise the Peer class
|
|
Peer.DEFAULT_PORTS = env.coin.PEER_DEFAULT_PORTS
|
|
self.env = env
|
|
self.db = db
|
|
|
|
# Our clearnet and Tor Peers, if any
|
|
sclass = env.coin.SESSIONCLS
|
|
self.myselves = [Peer(ident.host, sclass.server_features(env), 'env')
|
|
for ident in env.identities]
|
|
self.server_version_args = sclass.server_version_args()
|
|
# Peers have one entry per hostname. Once connected, the
|
|
# ip_addr property is either None, an onion peer, or the
|
|
# IP address that was connected to. Adding a peer will evict
|
|
# any other peers with the same host name or IP address.
|
|
self.peers: typing.Set[Peer] = set()
|
|
self.permit_onion_peer_time = time.time()
|
|
self.proxy = None
|
|
self.group = TaskGroup()
|
|
|
|
def _my_clearnet_peer(self):
|
|
"""Returns the clearnet peer representing this server, if any."""
|
|
clearnet = [peer for peer in self.myselves if not peer.is_tor]
|
|
return clearnet[0] if clearnet else None
|
|
|
|
def _set_peer_statuses(self):
|
|
"""Set peer statuses."""
|
|
cutoff = time.time() - STALE_SECS
|
|
for peer in self.peers:
|
|
if peer.bad:
|
|
peer.status = PEER_BAD
|
|
elif peer.last_good > cutoff:
|
|
peer.status = PEER_GOOD
|
|
elif peer.last_good:
|
|
peer.status = PEER_STALE
|
|
else:
|
|
peer.status = PEER_NEVER
|
|
|
|
def _features_to_register(self, peer, remote_peers):
|
|
"""If we should register ourselves to the remote peer, which has
|
|
reported the given list of known peers, return the clearnet
|
|
identity features to register, otherwise None.
|
|
"""
|
|
# Announce ourself if not present. Don't if disabled, we
|
|
# are a non-public IP address, or to ourselves.
|
|
if not self.env.peer_announce or peer in self.myselves:
|
|
return None
|
|
my = self._my_clearnet_peer()
|
|
if not my or not my.is_public:
|
|
return None
|
|
# Register if no matches, or ports have changed
|
|
for peer in my.matches(remote_peers):
|
|
if peer.tcp_port == my.tcp_port and peer.ssl_port == my.ssl_port:
|
|
return None
|
|
return my.features
|
|
|
|
def _permit_new_onion_peer(self):
|
|
"""Accept a new onion peer only once per random time interval."""
|
|
now = time.time()
|
|
if now < self.permit_onion_peer_time:
|
|
return False
|
|
self.permit_onion_peer_time = now + random.randrange(0, 1200)
|
|
return True
|
|
|
|
async def _import_peers(self):
|
|
"""Import hard-coded peers from a file or the coin defaults."""
|
|
imported_peers = self.myselves.copy()
|
|
# Add the hard-coded ones unless only reporting ourself
|
|
if self.env.peer_discovery != self.env.PD_SELF:
|
|
imported_peers.extend(Peer.from_real_name(real_name, 'coins.py')
|
|
for real_name in self.env.coin.PEERS)
|
|
await self._note_peers(imported_peers, limit=None)
|
|
|
|
async def _detect_proxy(self):
|
|
"""Detect a proxy if we don't have one and some time has passed since
|
|
the last attempt.
|
|
|
|
If found self.proxy is set to a SOCKSProxy instance, otherwise
|
|
None.
|
|
"""
|
|
host = self.env.tor_proxy_host
|
|
if self.env.tor_proxy_port is None:
|
|
ports = [9050, 9150, 1080]
|
|
else:
|
|
ports = [self.env.tor_proxy_port]
|
|
while True:
|
|
self.logger.info(f'trying to detect proxy on "{host}" '
|
|
f'ports {ports}')
|
|
proxy = await SOCKSProxy.auto_detect_host(host, ports, None)
|
|
if proxy:
|
|
self.proxy = proxy
|
|
self.logger.info(f'detected {proxy}')
|
|
return
|
|
self.logger.info('no proxy detected, will try later')
|
|
await sleep(900)
|
|
|
|
async def _note_peers(self, peers, limit=2, check_ports=False,
|
|
source=None):
|
|
"""Add a limited number of peers that are not already present."""
|
|
new_peers = []
|
|
for peer in peers:
|
|
if not peer.is_public or (peer.is_tor and not self.proxy):
|
|
continue
|
|
|
|
matches = peer.matches(self.peers)
|
|
if not matches:
|
|
new_peers.append(peer)
|
|
elif check_ports:
|
|
for match in matches:
|
|
if match.check_ports(peer):
|
|
self.logger.info(f'ports changed for {peer}')
|
|
match.retry_event.set()
|
|
|
|
if new_peers:
|
|
source = source or new_peers[0].source
|
|
if limit:
|
|
random.shuffle(new_peers)
|
|
use_peers = new_peers[:limit]
|
|
else:
|
|
use_peers = new_peers
|
|
for peer in use_peers:
|
|
self.logger.info(f'accepted new peer {peer} from {source}')
|
|
peer.retry_event = Event()
|
|
self.peers.add(peer)
|
|
await self.group.add(self._monitor_peer(peer))
|
|
|
|
async def _monitor_peer(self, peer):
|
|
# Stop monitoring if we were dropped (a duplicate peer)
|
|
while peer in self.peers:
|
|
if await self._should_drop_peer(peer):
|
|
self.peers.discard(peer)
|
|
break
|
|
# Figure out how long to sleep before retrying. Retry a
|
|
# good connection when it is about to turn stale, otherwise
|
|
# exponentially back off retries.
|
|
if peer.try_count == 0:
|
|
pause = STALE_SECS - WAKEUP_SECS * 2
|
|
else:
|
|
pause = WAKEUP_SECS * 2 ** peer.try_count
|
|
pending, done = await asyncio.wait([peer.retry_event.wait()], timeout=pause)
|
|
if done:
|
|
peer.retry_event.clear()
|
|
|
|
async def _should_drop_peer(self, peer):
|
|
peer.try_count += 1
|
|
is_good = False
|
|
for kind, port in peer.connection_port_pairs():
|
|
peer.last_try = time.time()
|
|
|
|
kwargs = {}
|
|
if kind == 'SSL':
|
|
kwargs['ssl'] = ssl.SSLContext(ssl.PROTOCOL_TLS)
|
|
|
|
host = self.env.cs_host(for_rpc=False)
|
|
if isinstance(host, list):
|
|
host = host[0]
|
|
|
|
if self.env.force_proxy or peer.is_tor:
|
|
if not self.proxy:
|
|
return
|
|
kwargs['proxy'] = self.proxy
|
|
kwargs['resolve'] = not peer.is_tor
|
|
elif host:
|
|
# Use our listening Host/IP for outgoing non-proxy
|
|
# connections so our peers see the correct source.
|
|
kwargs['local_addr'] = (host, None)
|
|
|
|
peer_text = f'[{peer}:{port} {kind}]'
|
|
try:
|
|
async with Connector(PeerSession, peer.host, port,
|
|
**kwargs) as session:
|
|
await asyncio.wait_for(
|
|
self._verify_peer(session, peer),
|
|
120 if peer.is_tor else 30
|
|
)
|
|
is_good = True
|
|
break
|
|
except BadPeerError as e:
|
|
self.logger.error(f'{peer_text} marking bad: ({e})')
|
|
peer.mark_bad()
|
|
break
|
|
except RPCError as e:
|
|
self.logger.error(f'{peer_text} RPC error: {e.message} '
|
|
f'({e.code})')
|
|
except (OSError, SOCKSError, ConnectionError, asyncio.TimeoutError) as e:
|
|
self.logger.info(f'{peer_text} {e}')
|
|
|
|
if is_good:
|
|
now = time.time()
|
|
elapsed = now - peer.last_try
|
|
self.logger.info(f'{peer_text} verified in {elapsed:.1f}s')
|
|
peer.try_count = 0
|
|
peer.last_good = now
|
|
peer.source = 'peer'
|
|
# At most 2 matches if we're a host name, potentially
|
|
# several if we're an IP address (several instances
|
|
# can share a NAT).
|
|
matches = peer.matches(self.peers)
|
|
for match in matches:
|
|
if match.ip_address:
|
|
if len(matches) > 1:
|
|
self.peers.remove(match)
|
|
# Force the peer's monitoring task to exit
|
|
match.retry_event.set()
|
|
elif peer.host in match.features['hosts']:
|
|
match.update_features_from_peer(peer)
|
|
else:
|
|
# Forget the peer if long-term unreachable
|
|
if peer.last_good and not peer.bad:
|
|
try_limit = 10
|
|
else:
|
|
try_limit = 3
|
|
if peer.try_count >= try_limit:
|
|
desc = 'bad' if peer.bad else 'unreachable'
|
|
self.logger.info(f'forgetting {desc} peer: {peer}')
|
|
return True
|
|
return False
|
|
|
|
async def _verify_peer(self, session, peer):
|
|
if not peer.is_tor:
|
|
address = session.peer_address()
|
|
if address:
|
|
peer.ip_addr = address[0]
|
|
|
|
# server.version goes first
|
|
message = 'server.version'
|
|
result = await session.send_request(message, self.server_version_args)
|
|
assert_good(message, result, list)
|
|
|
|
# Protocol version 1.1 returns a pair with the version first
|
|
if len(result) != 2 or not all(isinstance(x, str) for x in result):
|
|
raise BadPeerError(f'bad server.version result: {result}')
|
|
server_version, protocol_version = result
|
|
peer.server_version = server_version
|
|
peer.features['server_version'] = server_version
|
|
ptuple = protocol_tuple(protocol_version)
|
|
|
|
await asyncio.wait([
|
|
self._send_headers_subscribe(session, peer, ptuple),
|
|
self._send_server_features(session, peer),
|
|
self._send_peers_subscribe(session, peer)
|
|
])
|
|
|
|
async def _send_headers_subscribe(self, session, peer, ptuple):
|
|
message = 'blockchain.headers.subscribe'
|
|
result = await session.send_request(message)
|
|
assert_good(message, result, dict)
|
|
|
|
our_height = self.db.db_height
|
|
if ptuple < (1, 3):
|
|
their_height = result.get('block_height')
|
|
else:
|
|
their_height = result.get('height')
|
|
if not isinstance(their_height, int):
|
|
raise BadPeerError(f'invalid height {their_height}')
|
|
if abs(our_height - their_height) > 5:
|
|
raise BadPeerError(f'bad height {their_height:,d} '
|
|
f'(ours: {our_height:,d})')
|
|
|
|
# Check prior header too in case of hard fork.
|
|
check_height = min(our_height, their_height)
|
|
raw_header = await self.db.raw_header(check_height)
|
|
if ptuple >= (1, 4):
|
|
ours = raw_header.hex()
|
|
message = 'blockchain.block.header'
|
|
theirs = await session.send_request(message, [check_height])
|
|
assert_good(message, theirs, str)
|
|
if ours != theirs:
|
|
raise BadPeerError(f'our header {ours} and '
|
|
f'theirs {theirs} differ')
|
|
else:
|
|
ours = self.env.coin.electrum_header(raw_header, check_height)
|
|
ours = ours.get('prev_block_hash')
|
|
message = 'blockchain.block.get_header'
|
|
theirs = await session.send_request(message, [check_height])
|
|
assert_good(message, theirs, dict)
|
|
theirs = theirs.get('prev_block_hash')
|
|
if ours != theirs:
|
|
raise BadPeerError(f'our header hash {ours} and '
|
|
f'theirs {theirs} differ')
|
|
|
|
async def _send_server_features(self, session, peer):
|
|
message = 'server.features'
|
|
features = await session.send_request(message)
|
|
assert_good(message, features, dict)
|
|
hosts = [host.lower() for host in features.get('hosts', {})]
|
|
if self.env.coin.GENESIS_HASH != features.get('genesis_hash'):
|
|
raise BadPeerError('incorrect genesis hash')
|
|
elif peer.host.lower() in hosts:
|
|
peer.update_features(features)
|
|
else:
|
|
raise BadPeerError(f'not listed in own hosts list {hosts}')
|
|
|
|
async def _send_peers_subscribe(self, session, peer):
|
|
message = 'server.peers.subscribe'
|
|
raw_peers = await session.send_request(message)
|
|
assert_good(message, raw_peers, list)
|
|
|
|
# Check the peers list we got from a remote peer.
|
|
# Each is expected to be of the form:
|
|
# [ip_addr, hostname, ['v1.0', 't51001', 's51002']]
|
|
# Call add_peer if the remote doesn't appear to know about us.
|
|
try:
|
|
real_names = [' '.join([u[1]] + u[2]) for u in raw_peers]
|
|
peers = [Peer.from_real_name(real_name, str(peer))
|
|
for real_name in real_names]
|
|
except Exception:
|
|
raise BadPeerError('bad server.peers.subscribe response')
|
|
|
|
await self._note_peers(peers)
|
|
features = self._features_to_register(peer, peers)
|
|
if not features:
|
|
return
|
|
self.logger.info(f'registering ourself with {peer}')
|
|
# We only care to wait for the response
|
|
await session.send_request('server.add_peer', [features])
|
|
|
|
#
|
|
# External interface
|
|
#
|
|
async def discover_peers(self):
|
|
"""Perform peer maintenance. This includes
|
|
|
|
1) Forgetting unreachable peers.
|
|
2) Verifying connectivity of new peers.
|
|
3) Retrying old peers at regular intervals.
|
|
"""
|
|
if self.env.peer_discovery != self.env.PD_ON:
|
|
self.logger.info('peer discovery is disabled')
|
|
return
|
|
|
|
self.logger.info(f'beginning peer discovery. Force use of '
|
|
f'proxy: {self.env.force_proxy}')
|
|
|
|
self.group.add(self._detect_proxy())
|
|
self.group.add(self._import_peers())
|
|
|
|
def info(self) -> typing.Dict[str, int]:
|
|
"""The number of peers."""
|
|
self._set_peer_statuses()
|
|
counter = Counter(peer.status for peer in self.peers)
|
|
return {
|
|
'bad': counter[PEER_BAD],
|
|
'good': counter[PEER_GOOD],
|
|
'never': counter[PEER_NEVER],
|
|
'stale': counter[PEER_STALE],
|
|
'total': len(self.peers),
|
|
}
|
|
|
|
async def add_localRPC_peer(self, real_name):
|
|
"""Add a peer passed by the admin over LocalRPC."""
|
|
await self._note_peers([Peer.from_real_name(real_name, 'RPC')])
|
|
|
|
async def on_add_peer(self, features, source_info):
|
|
"""Add a peer (but only if the peer resolves to the source)."""
|
|
if not source_info:
|
|
self.logger.info('ignored add_peer request: no source info')
|
|
return False
|
|
source = source_info[0]
|
|
peers = Peer.peers_from_features(features, source)
|
|
if not peers:
|
|
self.logger.info('ignored add_peer request: no peers given')
|
|
return False
|
|
|
|
# Just look at the first peer, require it
|
|
peer = peers[0]
|
|
host = peer.host
|
|
if peer.is_tor:
|
|
permit = self._permit_new_onion_peer()
|
|
reason = 'rate limiting'
|
|
else:
|
|
getaddrinfo = asyncio.get_event_loop().getaddrinfo
|
|
try:
|
|
infos = await getaddrinfo(host, 80, type=socket.SOCK_STREAM)
|
|
except socket.gaierror:
|
|
permit = False
|
|
reason = 'address resolution failure'
|
|
else:
|
|
permit = any(source == info[-1][0] for info in infos)
|
|
reason = 'source-destination mismatch'
|
|
|
|
if permit:
|
|
self.logger.info(f'accepted add_peer request from {source} '
|
|
f'for {host}')
|
|
await self._note_peers([peer], check_ports=True)
|
|
else:
|
|
self.logger.warning(f'rejected add_peer request from {source} '
|
|
f'for {host} ({reason})')
|
|
|
|
return permit
|
|
|
|
def on_peers_subscribe(self, is_tor):
|
|
"""Returns the server peers as a list of (ip, host, details) tuples.
|
|
|
|
We return all peers we've connected to in the last day.
|
|
Additionally, if we don't have onion routing, we return a few
|
|
hard-coded onion servers.
|
|
"""
|
|
cutoff = time.time() - STALE_SECS
|
|
recent = [peer for peer in self.peers
|
|
if peer.last_good > cutoff and
|
|
not peer.bad and peer.is_public]
|
|
onion_peers = []
|
|
|
|
# Always report ourselves if valid (even if not public)
|
|
peers = {myself for myself in self.myselves
|
|
if myself.last_good > cutoff}
|
|
|
|
# Bucket the clearnet peers and select up to two from each
|
|
buckets = defaultdict(list)
|
|
for peer in recent:
|
|
if peer.is_tor:
|
|
onion_peers.append(peer)
|
|
else:
|
|
buckets[peer.bucket()].append(peer)
|
|
for bucket_peers in buckets.values():
|
|
random.shuffle(bucket_peers)
|
|
peers.update(bucket_peers[:2])
|
|
|
|
# Add up to 20% onion peers (but up to 10 is OK anyway)
|
|
random.shuffle(onion_peers)
|
|
max_onion = 50 if is_tor else max(10, len(peers) // 4)
|
|
|
|
peers.update(onion_peers[:max_onion])
|
|
|
|
return [peer.to_tuple() for peer in peers]
|
|
|
|
def proxy_peername(self):
|
|
"""Return the peername of the proxy, if there is a proxy, otherwise
|
|
None."""
|
|
return self.proxy.peername if self.proxy else None
|
|
|
|
def rpc_data(self):
|
|
"""Peer data for the peers RPC method."""
|
|
self._set_peer_statuses()
|
|
descs = ['good', 'stale', 'never', 'bad']
|
|
|
|
def peer_data(peer):
|
|
data = peer.serialize()
|
|
data['status'] = descs[peer.status]
|
|
return data
|
|
|
|
def peer_key(peer):
|
|
return (peer.bad, -peer.last_good)
|
|
|
|
return [peer_data(peer) for peer in sorted(self.peers, key=peer_key)]
|