lbry-sdk/scripts/seed_node.py

229 lines
8.9 KiB
Python
Raw Normal View History

import struct
2018-07-21 20:12:29 +02:00
import json
import logging
import argparse
import hashlib
from copy import deepcopy
from urllib import urlopen
2018-05-29 22:23:23 +02:00
from twisted.internet.epollreactor import install as install_epoll
install_epoll()
from twisted.internet import reactor, defer
from twisted.web import resource
from twisted.web.server import Site
from lbrynet import conf
from lbrynet.dht import constants
from lbrynet.dht.node import Node
from lbrynet.dht.error import TransportNotConnected
from lbrynet.core.log_support import configure_console, configure_twisted
from lbrynet.daemon.auth.server import AuthJSONRPCServer
# configure_twisted()
conf.initialize_settings()
configure_console()
lbrynet_handler = logging.getLogger("lbrynet").handlers[0]
log = logging.getLogger("dht router")
log.addHandler(lbrynet_handler)
log.setLevel(logging.INFO)
def node_id_supplier(seed="jack.lbry.tech"): # simple deterministic node id generator
h = hashlib.sha384()
h.update(seed)
while True:
next_id = h.digest()
yield next_id
h = hashlib.sha384()
h.update(seed)
h.update(next_id)
def get_external_ip():
response = json.loads(urlopen("https://api.lbry.io/ip").read())
if not response['success']:
raise ValueError("failed to get external ip")
return response['data']['ip']
def format_contact(contact):
return {
"node_id": contact.id.encode('hex'),
"address": contact.address,
2018-05-31 18:01:07 +02:00
"nodePort": contact.port,
"lastReplied": contact.lastReplied,
"lastRequested": contact.lastRequested,
2018-05-31 18:01:07 +02:00
"failedRPCs": contact.failedRPCs,
"lastFailed": None if not contact.failures else contact.failures[-1]
}
2018-05-31 18:01:07 +02:00
def format_datastore(node):
datastore = deepcopy(node._dataStore._dict)
result = {}
for key, values in datastore.items():
2018-05-31 18:01:07 +02:00
contacts = []
for (contact, value, last_published, originally_published, original_publisher_id) in values:
contact_dict = format_contact(contact)
contact_dict['peerPort'] = struct.unpack('>H', value[4:6])[0]
contact_dict['lastPublished'] = last_published
contact_dict['originallyPublished'] = originally_published
contact_dict['originalPublisherID'] = original_publisher_id.encode('hex')
contacts.append(contact_dict)
result[key.encode('hex')] = contacts
return result
class MultiSeedRPCServer(AuthJSONRPCServer):
2018-05-29 22:23:23 +02:00
def __init__(self, starting_node_port, nodes, rpc_port):
AuthJSONRPCServer.__init__(self, False)
self.port = None
self.rpc_port = rpc_port
self.external_ip = get_external_ip()
node_id_gen = node_id_supplier()
self._nodes = [Node(node_id=next(node_id_gen), udpPort=starting_node_port+i, externalIP=self.external_ip)
for i in range(nodes)]
self._own_addresses = [(self.external_ip, starting_node_port+i) for i in range(nodes)]
reactor.addSystemEventTrigger('after', 'startup', self.start)
@defer.inlineCallbacks
def start(self):
self.announced_startup = True
root = resource.Resource()
root.putChild('', self)
self.port = reactor.listenTCP(self.rpc_port, Site(root), interface='localhost')
log.info("starting %i nodes on %s, rpc available on localhost:%i", len(self._nodes), self.external_ip, self.rpc_port)
for node in self._nodes:
node.start_listening()
yield node._protocol._listening
for node1 in self._nodes:
for node2 in self._nodes:
if node1 is node2:
continue
try:
yield node1.addContact(node1.contact_manager.make_contact(node2.node_id, node2.externalIP,
node2.port, node1._protocol))
except TransportNotConnected:
pass
node1.safe_start_looping_call(node1._change_token_lc, constants.tokenSecretChangeInterval)
node1.safe_start_looping_call(node1._refresh_node_lc, constants.checkRefreshInterval)
node1._join_deferred = defer.succeed(True)
reactor.addSystemEventTrigger('before', 'shutdown', self.stop)
log.info("finished bootstrapping the network, running %i nodes", len(self._nodes))
@defer.inlineCallbacks
def stop(self):
yield self.port.stopListening()
yield defer.DeferredList([node.stop() for node in self._nodes])
def jsonrpc_get_node_ids(self):
return defer.succeed([node.node_id.encode('hex') for node in self._nodes])
def jsonrpc_node_datastore(self, node_id):
for node in self._nodes:
if node.node_id == node_id.decode('hex'):
return defer.succeed(format_datastore(node))
2018-05-31 18:01:07 +02:00
def jsonrpc_get_nodes_who_stored(self, blob_hash):
storing_nodes = {}
for node in self._nodes:
datastore = format_datastore(node)
if blob_hash in datastore:
storing_nodes[node.node_id.encode('hex')] = datastore[blob_hash]
return defer.succeed(storing_nodes)
def jsonrpc_node_routing_table(self, node_id):
def format_bucket(bucket):
return {
"contacts": [format_contact(contact) for contact in bucket._contacts],
"lastAccessed": bucket.lastAccessed
}
def format_routing(node):
return {
i: format_bucket(bucket) for i, bucket in enumerate(node._routingTable._buckets)
}
for node in self._nodes:
if node.node_id == node_id.decode('hex'):
return defer.succeed(format_routing(node))
def jsonrpc_restart_node(self, node_id):
for node in self._nodes:
if node.node_id == node_id.decode('hex'):
d = node.stop()
d.addCallback(lambda _: node.start(self._own_addresses))
return d
@defer.inlineCallbacks
def jsonrpc_local_node_rpc(self, from_node, query, args=()):
def format_result(response):
if isinstance(response, list):
return [[node_id.encode('hex'), address, port] for (node_id, address, port) in response]
if isinstance(response, dict):
return {'token': response['token'].encode('hex'), 'contacts': format_result(response['contacts'])}
return response
for node in self._nodes:
if node.node_id == from_node.decode('hex'):
fn = getattr(node, query)
self_contact = node.contact_manager.make_contact(node.node_id, node.externalIP, node.port, node._protocol)
if args:
args = (str(arg) if isinstance(arg, (str, unicode)) else int(arg) for arg in args)
result = yield fn(self_contact, *args)
else:
result = yield fn()
# print "result: %s" % result
defer.returnValue(format_result(result))
@defer.inlineCallbacks
def jsonrpc_node_rpc(self, from_node, to_node, query, args=()):
def format_result(response):
if isinstance(response, list):
return [[node_id.encode('hex'), address, port] for (node_id, address, port) in response]
if isinstance(response, dict):
return {'token': response['token'].encode('hex'), 'contacts': format_result(response['contacts'])}
return response
for node in self._nodes:
if node.node_id == from_node.decode('hex'):
remote = node._routingTable.getContact(to_node.decode('hex'))
fn = getattr(remote, query)
if args:
args = (str(arg).decode('hex') for arg in args)
result = yield fn(*args)
else:
result = yield fn()
defer.returnValue(format_result(result))
@defer.inlineCallbacks
def jsonrpc_get_nodes_who_know(self, ip_address):
nodes = []
for node_id in [n.node_id.encode('hex') for n in self._nodes]:
routing_info = yield self.jsonrpc_node_routing_table(node_id=node_id)
for index, bucket in routing_info.items():
if ip_address in map(lambda c: c['address'], bucket['contacts']):
nodes.append(node_id)
break
defer.returnValue(nodes)
def jsonrpc_node_status(self):
return defer.succeed({
node.node_id.encode('hex'): node._join_deferred is not None and node._join_deferred.called
for node in self._nodes
})
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--rpc_port', default=5280)
parser.add_argument('--starting_port', default=4455)
2018-05-29 22:23:23 +02:00
parser.add_argument('--nodes', default=32)
args = parser.parse_args()
MultiSeedRPCServer(int(args.starting_port), int(args.nodes), int(args.rpc_port))
reactor.run()
if __name__ == "__main__":
main()