783 lines
31 KiB
Python
Executable File
783 lines
31 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
'''
|
|
Onionr - P2P Microblogging Platform & Social network.
|
|
|
|
This file contains both the OnionrCommunicate class for communcating with peers
|
|
and code to operate as a daemon, getting commands from the command queue database (see core.Core.daemonQueue)
|
|
'''
|
|
'''
|
|
This program is free software: you can redistribute it and/or modify
|
|
it under the terms of the GNU General Public License as published by
|
|
the Free Software Foundation, either version 3 of the License, or
|
|
(at your option) any later version.
|
|
|
|
This program is distributed in the hope that it will be useful,
|
|
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
GNU General Public License for more details.
|
|
|
|
You should have received a copy of the GNU General Public License
|
|
along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
'''
|
|
import sqlite3, requests, hmac, hashlib, time, sys, os, math, logger, urllib.parse, base64, binascii, random, json, threading
|
|
import core, onionrutils, onionrcrypto, netcontroller, onionrproofs, btc, config, onionrplugins as plugins
|
|
|
|
class OnionrCommunicate:
|
|
def __init__(self, debug, developmentMode):
|
|
'''
|
|
OnionrCommunicate
|
|
|
|
This class handles communication with nodes in the Onionr network.
|
|
'''
|
|
|
|
self._core = core.Core()
|
|
self._utils = onionrutils.OnionrUtils(self._core)
|
|
self._crypto = onionrcrypto.OnionrCrypto(self._core)
|
|
self._netController = netcontroller.NetController(0) # arg is the HS port but not needed rn in this file
|
|
|
|
self.newHashes = {} # use this to not keep hashes around too long if we cant get their data
|
|
self.keepNewHash = 12
|
|
self.ignoredHashes = []
|
|
|
|
self.highFailureAmount = 7
|
|
|
|
self.communicatorThreads = 0
|
|
self.maxThreads = 75
|
|
self.processBlocksThreads = 0
|
|
self.lookupBlocksThreads = 0
|
|
|
|
self.blocksProcessing = [] # list of blocks currently processing, to avoid trying a block twice at once in 2 seperate threads
|
|
self.peerStatus = {} # network actions (active requests) for peers used mainly to prevent conflicting actions in threads
|
|
'''
|
|
logger.info('Starting Bitcoin Node... with Tor socks port:' + str(sys.argv[2]), timestamp=True)
|
|
try:
|
|
self.bitcoin = btc.OnionrBTC(torP=int(sys.argv[2]))
|
|
except _gdbm.error:
|
|
pass
|
|
logger.info('Bitcoin Node started, on block: ' + self.bitcoin.node.getBlockHash(self.bitcoin.node.getLastBlockHeight()), timestamp=True)
|
|
'''
|
|
#except:
|
|
#logger.fatal('Failed to start Bitcoin Node, exiting...')
|
|
#exit(1)
|
|
|
|
blockProcessTimer = 19
|
|
blockProcessAmount = 20
|
|
highFailureTimer = 0
|
|
highFailureRate = 10
|
|
heartBeatTimer = 0
|
|
heartBeatRate = 10
|
|
pexTimer = 120 # How often we should check for new peers
|
|
pexCount = 0
|
|
logger.debug('Communicator debugging enabled.')
|
|
torID = open('data/hs/hostname').read()
|
|
|
|
apiRunningCheckRate = 10
|
|
apiRunningCheckCount = 0
|
|
|
|
self.peerData = {} # Session data for peers (recent reachability, speed, etc)
|
|
|
|
if os.path.exists(self._core.queueDB):
|
|
self._core.clearDaemonQueue()
|
|
|
|
# Loads in and starts the enabled plugins
|
|
plugins.reload()
|
|
|
|
while True:
|
|
command = self._core.daemonQueue()
|
|
# Process blocks based on a timer
|
|
blockProcessTimer += 1
|
|
heartBeatTimer += 1
|
|
pexCount += 1
|
|
if highFailureTimer == highFailureRate:
|
|
highFailureTimer = 0
|
|
for i in self.peerData:
|
|
if self.peerData[i]['failCount'] >= self.highFailureAmount:
|
|
self.peerData[i]['failCount'] -= 1
|
|
if pexTimer == pexCount:
|
|
pT1 = threading.Thread(target=self.getNewPeers, name="pT1")
|
|
pT1.start()
|
|
pT2 = threading.Thread(target=self.getNewPeers, name="pT2")
|
|
pT2.start()
|
|
pexCount = 0 # TODO: do not reset timer if low peer count
|
|
if heartBeatRate == heartBeatTimer:
|
|
logger.debug('Communicator heartbeat')
|
|
heartBeatTimer = 0
|
|
if blockProcessTimer == blockProcessAmount:
|
|
lT1 = threading.Thread(target=self.lookupBlocks, name="lt1", args=(True,))
|
|
lT2 = threading.Thread(target=self.lookupBlocks, name="lt2", args=(True,))
|
|
lT3 = threading.Thread(target=self.lookupBlocks, name="lt3", args=(True,))
|
|
lT4 = threading.Thread(target=self.lookupBlocks, name="lt4", args=(True,))
|
|
pbT1 = threading.Thread(target=self.processBlocks, name='pbT1', args=(True,))
|
|
pbT2 = threading.Thread(target=self.processBlocks, name='pbT2', args=(True,))
|
|
pbT3 = threading.Thread(target=self.processBlocks, name='pbT3', args=(True,))
|
|
pbT4 = threading.Thread(target=self.processBlocks, name='pbT4', args=(True,))
|
|
if (self.maxThreads - 8) >= threading.active_count():
|
|
lT1.start()
|
|
lT2.start()
|
|
lT3.start()
|
|
lT4.start()
|
|
pbT1.start()
|
|
pbT2.start()
|
|
pbT3.start()
|
|
pbT4.start()
|
|
blockProcessTimer = 0
|
|
else:
|
|
logger.debug(threading.active_count())
|
|
logger.debug('Too many threads.')
|
|
if command != False:
|
|
if command[0] == 'shutdown':
|
|
logger.info('Daemon received exit command.', timestamp=True)
|
|
break
|
|
elif command[0] == 'announceNode':
|
|
announceAttempts = 3
|
|
announceAttemptCount = 0
|
|
announceVal = False
|
|
logger.info('Announcing node to ' + command[1], timestamp=True)
|
|
while not announceVal:
|
|
announceAttemptCount += 1
|
|
announceVal = self.performGet('announce', command[1], data=self._core.hsAdder.replace('\n', ''), skipHighFailureAddress=True)
|
|
logger.info(announceVal)
|
|
if announceAttemptCount >= announceAttempts:
|
|
logger.warn('Unable to announce to ' + command[1])
|
|
break
|
|
elif command[0] == 'runCheck':
|
|
logger.info('Status check; looks good.')
|
|
open('data/.runcheck', 'w+').close()
|
|
elif command[0] == 'event':
|
|
# todo
|
|
pass
|
|
elif command[0] == 'checkCallbacks':
|
|
try:
|
|
data = json.loads(command[1])
|
|
|
|
logger.info('Checking for callbacks with connection %s...' % data['id'])
|
|
|
|
self.check_callbacks(data, config.get('dc_execcallbacks', True))
|
|
|
|
events.event('incoming_direct_connection', data = {'callback' : True, 'communicator' : self, 'data' : data})
|
|
except Exception as e:
|
|
logger.error('Failed to interpret callbacks for checking', e)
|
|
elif command[0] == 'incomingDirectConnection':
|
|
try:
|
|
data = json.loads(command[1])
|
|
|
|
logger.info('Handling incoming connection %s...' % data['id'])
|
|
|
|
self.incoming_direct_connection(data)
|
|
|
|
events.event('incoming_direct_connection', data = {'callback' : False, 'communicator' : self, 'data' : data})
|
|
except Exception as e:
|
|
logger.error('Failed to handle callbacks for checking', e)
|
|
|
|
apiRunningCheckCount += 1
|
|
|
|
# check if local API is up
|
|
if apiRunningCheckCount > apiRunningCheckRate:
|
|
if self._core._utils.localCommand('ping') != 'pong':
|
|
for i in range(4):
|
|
if self._utils.localCommand('ping') == 'pong':
|
|
apiRunningCheckCount = 0
|
|
break # break for loop
|
|
time.sleep(1)
|
|
else:
|
|
# This executes if the api is NOT detected to be running
|
|
logger.error('Daemon detected API crash (or otherwise unable to reach API after long time), stopping...')
|
|
break # break main daemon loop
|
|
apiRunningCheckCount = 0
|
|
|
|
time.sleep(1)
|
|
|
|
self._netController.killTor()
|
|
return
|
|
|
|
future_callbacks = {}
|
|
connection_handlers = {}
|
|
id_peer_cache = {}
|
|
|
|
def get_connection_handlers(self, name = None):
|
|
'''
|
|
Returns a list of callback handlers by name, or, if name is None, it returns all handlers.
|
|
'''
|
|
|
|
if name is None:
|
|
return self.connection_handlers
|
|
elif name in self.connection_handlers:
|
|
return self.connection_handlers[name]
|
|
else:
|
|
return list()
|
|
|
|
def add_connection_handler(self, name, handler):
|
|
'''
|
|
Adds a function to be called when an connection that is NOT a callback is received.
|
|
Takes in the name of the communication type and the handler as input
|
|
'''
|
|
|
|
if not name in self.connection_handlers:
|
|
self.connection_handlers[name] = list()
|
|
|
|
self.connection_handlers[name].append(handler)
|
|
|
|
return
|
|
|
|
def remove_connection_handler(self, name, handler = None):
|
|
'''
|
|
Removes a connection handler if specified, or removes all by name
|
|
'''
|
|
|
|
if handler is None:
|
|
if name in self.connection_handlers:
|
|
self.connection_handlers[name].remove(handler)
|
|
elif name in self.connection_handlers:
|
|
del self.connection_handlers[name]
|
|
|
|
return
|
|
|
|
|
|
def set_callback(self, identifier, callback):
|
|
'''
|
|
(Over)writes a callback by communication identifier
|
|
'''
|
|
|
|
if not callback is None:
|
|
self.future_callbacks[identifier] = callback
|
|
return True
|
|
|
|
return False
|
|
|
|
def unset_callback(self, identifier):
|
|
'''
|
|
Unsets a callback by communication identifier, if set
|
|
'''
|
|
|
|
if identifier in future_callbacks:
|
|
del self.future_callbacks[identifier]
|
|
return True
|
|
|
|
return False
|
|
|
|
def get_callback(self, identifier):
|
|
'''
|
|
Returns a callback by communication identifier if set, or None
|
|
'''
|
|
|
|
if identifier in self.future_callbacks:
|
|
return self.future_callbacks[id]
|
|
|
|
return None
|
|
|
|
def direct_connect(self, peer, data = None, callback = None, log = True):
|
|
'''
|
|
Communicates something directly with the client
|
|
|
|
- `peer` should obviously be the peer id to request.
|
|
- `data` should be a dict (NOT str), with the parameter "type"
|
|
ex. {'type': 'sendMessage', 'content': 'hey, this is a dm'}
|
|
In that dict, the key 'token' must NEVER be set. If it is, it will
|
|
be overwritten.
|
|
- if `callback` is set to a function, it will call that function
|
|
back if/when the client the request is sent to decides to respond.
|
|
Do NOT depend on a response, because users can configure their
|
|
clients not to respond to this type of request.
|
|
- `log` is set to True by default-- what this does is log the
|
|
request for debug purposes. Should be False for sensitive actions.
|
|
'''
|
|
|
|
# TODO: Timing attack prevention
|
|
try:
|
|
# does not need to be secure random, only used for keeping track of async responses
|
|
# Actually, on second thought, it does need to be secure random. Otherwise, if it is predictable, someone could trigger arbitrary callbacks that have been saved on the local node, wrecking all kinds of havoc. Better just to keep it secure random.
|
|
identifier = self._utils.token(32)
|
|
if 'id' in data:
|
|
identifier = data['id']
|
|
|
|
if not identifier in id_peer_cache:
|
|
id_peer_cache[identifier] = peer
|
|
|
|
if type(data) == str:
|
|
# if someone inputs a string instead of a dict, it will assume it's the type
|
|
data = {'type' : data}
|
|
|
|
data['id'] = identifier
|
|
data['token'] = '' # later put PoW stuff here or whatever is needed
|
|
data_str = json.dumps(data)
|
|
|
|
events.event('outgoing_direct_connection', data = {'callback' : True, 'communicator' : self, 'data' : data, 'id' : identifier, 'token' : token, 'peer' : peer, 'callback' : callback, 'log' : log})
|
|
|
|
logger.debug('Direct connection (identifier: "%s"): %s' % (identifier, data_str))
|
|
try:
|
|
self.performGet('directMessage', peer, data_str)
|
|
except:
|
|
logger.warn('Failed to connect to peer: "%s".' % str(peer))
|
|
return False
|
|
|
|
if not callback is None:
|
|
self.set_callback(identifier, callback)
|
|
|
|
return True
|
|
except Exception as e:
|
|
logger.warn('Unknown error, failed to execute direct connect (peer: "%s").' % str(peer), e)
|
|
|
|
return False
|
|
|
|
def direct_connect_response(self, identifier, data, peer = None, callback = None, log = True):
|
|
'''
|
|
Responds to a previous connection. Hostname will be pulled from id_peer_cache if not specified in `peer` parameter.
|
|
|
|
If yet another callback is requested, it can be put in the `callback` parameter.
|
|
'''
|
|
|
|
if config.get('dc_response', True):
|
|
data['id'] = identifier
|
|
data['sender'] = open('data/hs/hostname').read()
|
|
data['callback'] = True
|
|
|
|
if (origin is None) and (identifier in id_peer_cache):
|
|
origin = id_peer_cache[identifier]
|
|
|
|
if not identifier in id_peer_cache:
|
|
id_peer_cache[identifier] = peer
|
|
|
|
if origin is None:
|
|
logger.warn('Failed to identify peer for connection %s' % str(identifier))
|
|
return False
|
|
else:
|
|
return self.direct_connect(peer, data = data, callback = callback, log = log)
|
|
else:
|
|
logger.warn('Node tried to respond to direct connection id %s, but it was rejected due to `dc_response` restriction.' % str(identifier))
|
|
return False
|
|
|
|
|
|
def check_callbacks(self, data, execute = True, remove = True):
|
|
'''
|
|
Check if a callback is set, and if so, execute it
|
|
'''
|
|
|
|
try:
|
|
if type(data) is str:
|
|
data = json.loads(data)
|
|
|
|
if 'id' in data: # TODO: prevent enumeration, require extra PoW
|
|
identifier = data['id']
|
|
|
|
if identifier in self.future_callbacks:
|
|
if execute:
|
|
self.get_callback(identifier)(data)
|
|
logger.debug('Request callback "%s" executed.' % str(identifier))
|
|
if remove:
|
|
self.unset_callback(identifier)
|
|
|
|
return True
|
|
|
|
logger.warn('Unable to find request callback for ID "%s".' % str(identifier))
|
|
else:
|
|
logger.warn('Unable to identify callback request, `id` parameter missing: %s' % json.dumps(data))
|
|
except Exception as e:
|
|
logger.warn('Unknown error, failed to execute direct connection callback (peer: "%s").' % str(peer), e)
|
|
|
|
return False
|
|
|
|
def incoming_direct_connection(self, data):
|
|
'''
|
|
This code is run whenever there is a new incoming connection.
|
|
'''
|
|
|
|
if 'type' in data and data['type'] in self.connection_handlers:
|
|
for handler in self.get_connection_handlers(name):
|
|
handler(data)
|
|
|
|
return
|
|
|
|
def getNewPeers(self):
|
|
'''
|
|
Get new peers and ed25519 keys
|
|
'''
|
|
|
|
peersCheck = 1 # Amount of peers to ask for new peers + keys
|
|
peersChecked = 0
|
|
peerList = list(self._core.listAdders()) # random ordered list of peers
|
|
newKeys = []
|
|
newAdders = []
|
|
if len(peerList) > 0:
|
|
maxN = len(peerList) - 1
|
|
else:
|
|
peersCheck = 0
|
|
maxN = 0
|
|
|
|
if len(peerList) > peersCheck:
|
|
peersCheck = len(peerList)
|
|
|
|
while peersCheck > peersChecked:
|
|
#i = secrets.randbelow(maxN) # cant use prior to 3.6
|
|
i = random.randint(0, maxN)
|
|
|
|
try:
|
|
if self.peerStatusTaken(peerList[i], 'pex') or self.peerStatusTaken(peerList[i], 'kex'):
|
|
continue
|
|
except IndexError:
|
|
pass
|
|
|
|
logger.info('Using %s to find new peers...' % peerList[i], timestamp=True)
|
|
|
|
try:
|
|
newAdders = self.performGet('pex', peerList[i], skipHighFailureAddress=True)
|
|
if not newAdders is False: # keep the is False thing in there, it might not be bool
|
|
logger.debug('Attempting to merge address: %s' % str(newAdders))
|
|
self._utils.mergeAdders(newAdders)
|
|
except requests.exceptions.ConnectionError:
|
|
logger.info('%s connection failed' % peerList[i], timestamp=True)
|
|
continue
|
|
else:
|
|
try:
|
|
logger.info('Using %s to find new keys...' % peerList[i])
|
|
newKeys = self.performGet('kex', peerList[i], skipHighFailureAddress=True)
|
|
logger.debug('Attempting to merge pubkey: %s' % str(newKeys))
|
|
# TODO: Require keys to come with POW token (very large amount of POW)
|
|
self._utils.mergeKeys(newKeys)
|
|
except requests.exceptions.ConnectionError:
|
|
logger.info('%s connection failed' % peerList[i], timestamp=True)
|
|
continue
|
|
else:
|
|
peersChecked += 1
|
|
return
|
|
|
|
def lookupBlocks(self, isThread=False):
|
|
'''
|
|
Lookup blocks and merge new ones
|
|
'''
|
|
if isThread:
|
|
self.lookupBlocksThreads += 1
|
|
peerList = self._core.listAdders()
|
|
blocks = ''
|
|
|
|
for i in peerList:
|
|
if self.peerStatusTaken(i, 'getBlockHashes') or self.peerStatusTaken(i, 'getDBHash'):
|
|
continue
|
|
try:
|
|
if self.peerData[i]['failCount'] >= self.highFailureAmount:
|
|
continue
|
|
except KeyError:
|
|
pass
|
|
|
|
lastDB = self._core.getAddressInfo(i, 'DBHash')
|
|
|
|
if lastDB == None:
|
|
logger.debug('Fetching hash from %s, no previous known.' % str(i))
|
|
else:
|
|
logger.debug('Fetching hash from %s, %s last known' % (str(i), str(lastDB)))
|
|
|
|
currentDB = self.performGet('getDBHash', i)
|
|
|
|
if currentDB != False:
|
|
logger.debug('%s hash db (from request): %s' % (str(i), str(currentDB)))
|
|
else:
|
|
logger.warn('Failed to get hash db status for %s' % str(i))
|
|
|
|
if currentDB != False:
|
|
if lastDB != currentDB:
|
|
logger.debug('Fetching hash from %s - %s current hash.' % (str(i), currentDB))
|
|
try:
|
|
blocks += self.performGet('getBlockHashes', i)
|
|
except TypeError:
|
|
logger.warn('Failed to get data hash from %s' % str(i))
|
|
self.peerData[i]['failCount'] -= 1
|
|
if self._utils.validateHash(currentDB):
|
|
self._core.setAddressInfo(i, "DBHash", currentDB)
|
|
|
|
if len(blocks.strip()) != 0:
|
|
pass
|
|
#logger.debug('BLOCKS:' + blocks)
|
|
|
|
blockList = blocks.split('\n')
|
|
|
|
for i in blockList:
|
|
if len(i.strip()) == 0:
|
|
continue
|
|
if self._utils.hasBlock(i):
|
|
continue
|
|
if i in self.ignoredHashes:
|
|
continue
|
|
|
|
#logger.debug('Exchanged block (blockList): ' + i)
|
|
if not self._utils.validateHash(i):
|
|
# skip hash if it isn't valid
|
|
logger.warn('Hash %s is not valid' % str(i))
|
|
continue
|
|
else:
|
|
self.newHashes[i] = 0
|
|
logger.debug('Adding %s to hash database...' % str(i))
|
|
self._core.addToBlockDB(i)
|
|
self.lookupBlocksThreads -= 1
|
|
return
|
|
|
|
def processBlocks(self, isThread=False):
|
|
'''
|
|
Work with the block database and download any missing blocks
|
|
|
|
This is meant to be called from the communicator daemon on its timer.
|
|
'''
|
|
if isThread:
|
|
self.processBlocksThreads += 1
|
|
for i in self._core.getBlockList(unsaved=True).split("\n"):
|
|
if i != "":
|
|
if i in self.blocksProcessing or i in self.ignoredHashes:
|
|
#logger.debug('already processing ' + i)
|
|
continue
|
|
else:
|
|
self.blocksProcessing.append(i)
|
|
try:
|
|
self.newHashes[i]
|
|
except KeyError:
|
|
self.newHashes[i] = 0
|
|
|
|
# check if a new hash has been around too long, delete it from database and add it to ignore list
|
|
if self.newHashes[i] >= self.keepNewHash:
|
|
logger.warn('Ignoring block %s because it took to long to get valid data.' % str(i))
|
|
del self.newHashes[i]
|
|
self._core.removeBlock(i)
|
|
self.ignoredHashes.append(i)
|
|
continue
|
|
|
|
self.newHashes[i] += 1
|
|
logger.warn('Block is unsaved: %s' % str(i))
|
|
data = self.downloadBlock(i)
|
|
|
|
# if block was successfull gotten (hash already verified)
|
|
if data:
|
|
del self.newHashes[i] # remove from probation list
|
|
|
|
# deal with block metadata
|
|
blockContent = self._core.getData(i)
|
|
try:
|
|
blockContent = blockContent.encode()
|
|
except AttributeError:
|
|
pass
|
|
try:
|
|
#blockMetadata = json.loads(self._core.getData(i)).split('}')[0] + '}'
|
|
blockMetadata = json.loads(blockContent[:blockContent.rfind(b'}') + 1].decode())
|
|
try:
|
|
blockMeta2 = json.loads(blockMetadata['meta'])
|
|
except KeyError:
|
|
blockMeta2 = {'type': ''}
|
|
pass
|
|
blockContent = blockContent[blockContent.rfind(b'}') + 1:]
|
|
try:
|
|
blockContent = blockContent.decode()
|
|
except AttributeError:
|
|
pass
|
|
|
|
if not self.verifyPow(blockContent, blockMeta2):
|
|
logger.warn("%s has invalid or insufficient proof of work token, deleting..." % str(i))
|
|
self._core.removeBlock(i)
|
|
continue
|
|
|
|
try:
|
|
blockMetadata['sig']
|
|
blockMeta2['id']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
#blockData = json.dumps(blockMetadata['meta']) + blockMetadata[blockMetadata.rfind(b'}') + 1:]
|
|
|
|
creator = self._utils.getPeerByHashId(blockMeta2['id'])
|
|
try:
|
|
creator = creator.decode()
|
|
except AttributeError:
|
|
pass
|
|
|
|
if self._core._crypto.edVerify(blockMetaData['meta'] + blockContent, creator, blockMetadata['sig'], encodedData=True):
|
|
logger.info('%s was signed' % str(i))
|
|
self._core.updateBlockInfo(i, 'sig', 'true')
|
|
else:
|
|
logger.warn('%s has an invalid signature' % str(i))
|
|
self._core.updateBlockInfo(i, 'sig', 'false')
|
|
try:
|
|
logger.info('Block type is %s' % str(blockMeta2['type']))
|
|
self._core.updateBlockInfo(i, 'dataType', blockMeta2['type'])
|
|
self.removeBlockFromProcessingList(i)
|
|
self.removeBlockFromProcessingList(i)
|
|
except KeyError:
|
|
logger.warn('Block has no type')
|
|
pass
|
|
except json.decoder.JSONDecodeError:
|
|
logger.warn('Could not decode block metadata')
|
|
self.removeBlockFromProcessingList(i)
|
|
self.processBlocksThreads -= 1
|
|
return
|
|
|
|
def removeBlockFromProcessingList(self, block):
|
|
try:
|
|
self.blocksProcessing.remove(block)
|
|
except ValueError:
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
def downloadBlock(self, hash, peerTries=3):
|
|
'''
|
|
Download a block from random order of peers
|
|
'''
|
|
|
|
retVal = False
|
|
peerList = self._core.listAdders()
|
|
blocks = ''
|
|
peerTryCount = 0
|
|
|
|
for i in peerList:
|
|
try:
|
|
if self.peerData[i]['failCount'] >= self.highFailureAmount:
|
|
continue
|
|
except KeyError:
|
|
pass
|
|
if peerTryCount >= peerTries:
|
|
break
|
|
|
|
hasher = hashlib.sha3_256()
|
|
data = self.performGet('getData', i, hash, skipHighFailureAddress=True)
|
|
|
|
if data == False or len(data) > 10000000 or data == '':
|
|
peerTryCount += 1
|
|
continue
|
|
|
|
try:
|
|
data = base64.b64decode(data)
|
|
except binascii.Error:
|
|
data = b''
|
|
|
|
hasher.update(data)
|
|
digest = hasher.hexdigest()
|
|
|
|
if type(digest) is bytes:
|
|
digest = digest.decode()
|
|
|
|
if digest == hash.strip():
|
|
self._core.setData(data)
|
|
logger.info('Successfully obtained data for %s' % str(hash), timestamp=True)
|
|
retVal = True
|
|
break
|
|
'''
|
|
if data.startswith(b'-txt-'):
|
|
self._core.setBlockType(hash, 'txt')
|
|
if len(data) < 120:
|
|
logger.debug('Block text:\n' + data.decode())
|
|
'''
|
|
else:
|
|
logger.warn("Failed to validate %s -- hash calculated was %s" % (hash, digest))
|
|
peerTryCount += 1
|
|
|
|
return retVal
|
|
|
|
def verifyPow(self, blockContent, metadata):
|
|
'''
|
|
Verifies the proof of work associated with a block
|
|
'''
|
|
retData = False
|
|
try:
|
|
metadata['powToken']
|
|
metadata['powHash']
|
|
token = metadata['powToken']
|
|
except KeyError:
|
|
return False
|
|
dataLen = len(blockContent)
|
|
|
|
expectedHash = self._crypto.blake2bHash(base64.b64decode(metadata['powToken']) + self._crypto.blake2bHash(blockContent.encode()))
|
|
difficulty = 0
|
|
try:
|
|
expectedHash = expectedHash.decode()
|
|
except AttributeError:
|
|
pass
|
|
if metadata['powHash'] == expectedHash:
|
|
difficulty = math.floor(dataLen/1000000)
|
|
|
|
mainHash = '0000000000000000000000000000000000000000000000000000000000000000'#nacl.hash.blake2b(nacl.utils.random()).decode()
|
|
puzzle = mainHash[0:difficulty]
|
|
|
|
if metadata['powHash'][0:difficulty] == puzzle:
|
|
logger.info('Validated block pow')
|
|
retData = True
|
|
else:
|
|
logger.warn("Invalid token (#1)")
|
|
else:
|
|
logger.warn('Invalid token (#2): Expected hash %s, got hash %s...' % (metadata['powHash'], expectedHash))
|
|
|
|
return retData
|
|
|
|
def urlencode(self, data):
|
|
'''
|
|
URL encodes the data
|
|
'''
|
|
return urllib.parse.quote_plus(data)
|
|
|
|
def performGet(self, action, peer, data=None, skipHighFailureAddress=False, peerType='tor', selfCheck=True):
|
|
'''
|
|
Performs a request to a peer through Tor or i2p (currently only Tor)
|
|
'''
|
|
|
|
if not peer.endswith('.onion') and not peer.endswith('.onion/'):
|
|
raise PeerError('Currently only Tor .onion peers are supported. You must manually specify .onion')
|
|
|
|
if len(self._core.hsAdder.strip()) == 0:
|
|
raise Exception("Could not perform self address check in performGet due to not knowing our address")
|
|
if selfCheck:
|
|
if peer.replace('/', '') == self._core.hsAdder:
|
|
logger.warn('Tried to performGet to own hidden service, but selfCheck was not set to false')
|
|
return
|
|
|
|
# Store peer in peerData dictionary (non permanent)
|
|
if not peer in self.peerData:
|
|
self.peerData[peer] = {'connectCount': 0, 'failCount': 0, 'lastConnectTime': math.floor(time.time())}
|
|
socksPort = sys.argv[2]
|
|
'''We use socks5h to use tor as DNS'''
|
|
proxies = {'http': 'socks5://127.0.0.1:' + str(socksPort), 'https': 'socks5://127.0.0.1:' + str(socksPort)}
|
|
headers = {'user-agent': 'PyOnionr'}
|
|
url = 'http://' + peer + '/public/?action=' + self.urlencode(action)
|
|
if data != None:
|
|
url = url + '&data=' + self.urlencode(data)
|
|
try:
|
|
if skipHighFailureAddress and self.peerData[peer]['failCount'] > self.highFailureAmount:
|
|
retData = False
|
|
logger.debug('Skipping %s because of high failure rate.' % peer)
|
|
else:
|
|
self.peerStatus[peer] = action
|
|
logger.debug('Contacting %s on port %s' % (peer, str(socksPort)))
|
|
r = requests.get(url, headers=headers, proxies=proxies, timeout=(15, 30))
|
|
retData = r.text
|
|
except requests.exceptions.RequestException as e:
|
|
logger.debug("%s failed with peer %s" % (action, peer))
|
|
retData = False
|
|
|
|
if not retData:
|
|
self.peerData[peer]['failCount'] += 1
|
|
else:
|
|
self.peerData[peer]['connectCount'] += 1
|
|
self.peerData[peer]['failCount'] -= 1
|
|
self.peerData[peer]['lastConnectTime'] = math.floor(time.time())
|
|
return retData
|
|
|
|
def peerStatusTaken(self, peer, status):
|
|
'''
|
|
Returns if a peer is currently performing a specified action
|
|
'''
|
|
try:
|
|
if self.peerStatus[peer] == status:
|
|
return True
|
|
except KeyError:
|
|
pass
|
|
return False
|
|
|
|
shouldRun = False
|
|
debug = True
|
|
developmentMode = False
|
|
if config.get('devmode', True):
|
|
developmentMode = True
|
|
try:
|
|
if sys.argv[1] == 'run':
|
|
shouldRun = True
|
|
except IndexError:
|
|
pass
|
|
if shouldRun:
|
|
try:
|
|
OnionrCommunicate(debug, developmentMode)
|
|
except KeyboardInterrupt:
|
|
sys.exit(1)
|
|
pass
|