Bitmessage/PyBitmessage-2025-01-22
Archived
2
0
Fork 0
Code Issues 167 Pull Requests 53 Actions Packages Projects Releases 5 Wiki Activity

Some cleanup in anticipation of IPV6.

Browse Source 
* Added message_parsers.py to separate parsing from protocol
  logic.
* Implemented addr parsing.
* Refactored receiveDataThread.recaddr.
* Added packNetworkAddress and unpackNetworkAddress in
  shared.py to put IPV6 address handling in one place.
* Have [,un]packNetworkAddress use inet_pton and inet_ntop.
* Convert some print statements to logger messages.
This commit is contained in:
Gregor Robinson 2013-07-18 13:00:55 +01:00
parent d93d923364
commit 90771f377c
5 changed files with 301 additions and 283 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/addresses.py
View File

@ -1,5 +1,6 @@
import hashlib import hashlib
from struct import * from struct import pack
from struct import unpack
from pyelliptic import arithmetic from pyelliptic import arithmetic

426
src/class_receiveDataThread.py
View File

@ -1,26 +1,28 @@
doTimingAttackMitigation = True doTimingAttackMitigation = True
# Libraries.
import hashlib
import pickle
from pyelliptic.openssl import OpenSSL
import random
import shared
import socket
import string
from struct import unpack, pack
from subprocess import call # used when the API must execute an outside program
import sys
import time import time
import threading import threading
import shared
import hashlib
import socket
import pickle
import random
from struct import unpack, pack
import sys
import string
from subprocess import call # used when the API must execute an outside program
from pyelliptic.openssl import OpenSSL
import highlevelcrypto # Project imports.
from addresses import * from addresses import *
import helper_generic from debug import logger
import helper_bitcoin import helper_bitcoin
import helper_generic
import helper_inbox import helper_inbox
import helper_sent import helper_sent
import highlevelcrypto
import tr import tr
#from bitmessagemain import shared.lengthOfTimeToLeaveObjectsInInventory, shared.lengthOfTimeToHoldOnToAllPubkeys, shared.maximumAgeOfAnObjectThatIAmWillingToAccept, shared.maximumAgeOfObjectsThatIAdvertiseToOthers, shared.maximumAgeOfNodesThatIAdvertiseToOthers, shared.numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer, shared.neededPubkeys
# This thread is created either by the synSenderThread(for outgoing # This thread is created either by the synSenderThread(for outgoing
# connections) or the singleListenerThread(for incoming connectiosn). # connections) or the singleListenerThread(for incoming connectiosn).
@ -61,8 +63,8 @@ class receiveDataThread(threading.Thread):
self.someObjectsOfWhichThisRemoteNodeIsAlreadyAware = someObjectsOfWhichThisRemoteNodeIsAlreadyAware self.someObjectsOfWhichThisRemoteNodeIsAlreadyAware = someObjectsOfWhichThisRemoteNodeIsAlreadyAware
def run(self): def run(self):
with shared.printLock: logger.info('ID of the receiveDataThread is %s. The size of the shared.connectedHostsList '
print 'ID of the receiveDataThread is', str(id(self)) + '. The size of the shared.connectedHostsList is now', len(shared.connectedHostsList) 'is now %s' % (id(self), len(shared.connectedHostsList)))
while True: while True:
dataLen = len(self.data) dataLen = len(self.data)
@ -70,7 +72,8 @@ class receiveDataThread(threading.Thread):
self.data += self.sock.recv(4096) self.data += self.sock.recv(4096)
except socket.timeout: except socket.timeout:
with shared.printLock: with shared.printLock:
print 'Timeout occurred waiting for data from', self.HOST + '. Closing receiveData thread. (ID:', str(id(self)) + ')' logger.info('Timeout occurred waiting for data from %s. Closing receiveData '
'thread. (ID: %s).' % (self.HOST, id(self)))
break break
except Exception as err: except Exception as err:
@ -78,18 +81,17 @@ class receiveDataThread(threading.Thread):
print 'sock.recv error. Closing receiveData thread (HOST:', self.HOST, 'ID:', str(id(self)) + ').', err print 'sock.recv error. Closing receiveData thread (HOST:', self.HOST, 'ID:', str(id(self)) + ').', err
break break
# print 'Received', repr(self.data) # logger.debug('Received', repr(self.data))
if len(self.data) == dataLen: # If self.sock.recv returned no data: if len(self.data) == dataLen: # If self.sock.recv returned no data:
with shared.printLock: logger.debug('Connection to %s closed. Closing receiveData thread. (ID: %s)',
print 'Connection to', self.HOST, 'closed. Closing receiveData thread. (ID:', str(id(self)) + ')' self.HOST, id(self))
break break
else: else:
self.processData() self.processData()
try: try:
del self.selfInitiatedConnections[self.streamNumber][self] del self.selfInitiatedConnections[self.streamNumber][self]
with shared.printLock: logger.info('removed self (a receiveDataThread) from selfInitiatedConnections')
print 'removed self (a receiveDataThread) from selfInitiatedConnections'
except: except:
pass pass
@ -97,8 +99,8 @@ class receiveDataThread(threading.Thread):
try: try:
del shared.connectedHostsList[self.HOST] del shared.connectedHostsList[self.HOST]
except Exception as err: except Exception as err:
with shared.printLock: logger.info('Could not delete %s from shared.connectedHostsList. %s',
print 'Could not delete', self.HOST, 'from shared.connectedHostsList.', err self.HOST, err)
try: try:
del shared.numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer[ del shared.numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer[
@ -106,21 +108,19 @@ class receiveDataThread(threading.Thread):
except: except:
pass pass
shared.UISignalQueue.put(('updateNetworkStatusTab', 'no data')) shared.UISignalQueue.put(('updateNetworkStatusTab', 'no data'))
with shared.printLock: logger.info('The size of the connectedHostsList is now: %s', len(shared.connectedHostsList))
print 'The size of the connectedHostsList is now:', len(shared.connectedHostsList)
def processData(self): def processData(self):
# if shared.verbose >= 3: # if shared.verbose >= 3:
# with shared.printLock: # logger.debug('self.data is currently ', repr(self.data))
# print 'self.data is currently ', repr(self.data)
# #
if len(self.data) < 20: # if so little of the data has arrived that we can't even unpack the payload length if len(self.data) < 20: # if so little of the data has arrived that we can't even unpack the payload length
return return
if self.data[0:4] != '\xe9\xbe\xb4\xd9': if self.data[0:4] != '\xe9\xbe\xb4\xd9':
if shared.verbose >= 1: if shared.verbose >= 1:
with shared.printLock: logger.info('The magic bytes were not correct. First 40 bytes of data: %s',
print 'The magic bytes were not correct. First 40 bytes of data: ' + repr(self.data[0:40]) repr(self.data[0:40]))
self.data = "" self.data = ""
return return
@ -128,7 +128,7 @@ class receiveDataThread(threading.Thread):
if len(self.data) < self.payloadLength + 24: # check if the whole message has arrived yet. if len(self.data) < self.payloadLength + 24: # check if the whole message has arrived yet.
return return
if self.data[20:24] != hashlib.sha512(self.data[24:self.payloadLength + 24]).digest()[0:4]: # test the checksum in the message. If it is correct... if self.data[20:24] != hashlib.sha512(self.data[24:self.payloadLength + 24]).digest()[0:4]: # test the checksum in the message. If it is correct...
print 'Checksum incorrect. Clearing this message.' logger.info('Checksum incorrect. Clearing this message.')
self.data = self.data[self.payloadLength + 24:] self.data = self.data[self.payloadLength + 24:]
self.processData() self.processData()
return return
@ -142,8 +142,8 @@ class receiveDataThread(threading.Thread):
shared.knownNodesLock.release() shared.knownNodesLock.release()
if self.payloadLength <= 180000000: # If the size of the message is greater than 180MB, ignore it. (I get memory errors when processing messages much larger than this though it is concievable that this value will have to be lowered if some systems are less tolarant of large messages.) if self.payloadLength <= 180000000: # If the size of the message is greater than 180MB, ignore it. (I get memory errors when processing messages much larger than this though it is concievable that this value will have to be lowered if some systems are less tolarant of large messages.)
remoteCommand = self.data[4:16] remoteCommand = self.data[4:16]
with shared.printLock: logger.info('remoteCommand %s from %s.',
print 'remoteCommand', repr(remoteCommand.replace('\x00', '')), ' from', self.HOST repr(remoteCommand.replace('\x00', '')), self.HOST)
if remoteCommand == 'version\x00\x00\x00\x00\x00': if remoteCommand == 'version\x00\x00\x00\x00\x00':
self.recversion(self.data[24:self.payloadLength + 24]) self.recversion(self.data[24:self.payloadLength + 24])
@ -178,15 +178,14 @@ class receiveDataThread(threading.Thread):
objectHash, = random.sample( objectHash, = random.sample(
self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave, 1) self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave, 1)
if objectHash in shared.inventory: if objectHash in shared.inventory:
with shared.printLock: logger.debug('Inventory (in memory) already has object listed in inv message.')
print 'Inventory (in memory) already has object listed in inv message.'
del self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave[ del self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave[
objectHash] objectHash]
elif shared.isInSqlInventory(objectHash): elif shared.isInSqlInventory(objectHash):
if shared.verbose >= 3: if shared.verbose >= 3:
with shared.printLock: logger.debug('Inventory (SQL on disk) already has object listed in inv '
print 'Inventory (SQL on disk) already has object listed in inv message.' 'message.')
del self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave[ del self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave[
objectHash] objectHash]
@ -195,8 +194,11 @@ class receiveDataThread(threading.Thread):
del self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave[ del self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave[
objectHash] # It is possible that the remote node doesn't respond with the object. In that case, we'll very likely get it from someone else anyway. objectHash] # It is possible that the remote node doesn't respond with the object. In that case, we'll very likely get it from someone else anyway.
if len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) == 0: if len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) == 0:
with shared.printLock: logger.debug('(concerning %s) number of '
print '(concerning', self.HOST + ')', 'number of objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave is now', len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) 'objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave '
'is now %s',
self.HOST,
len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave))
try: try:
del shared.numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer[ del shared.numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer[
@ -205,8 +207,11 @@ class receiveDataThread(threading.Thread):
pass pass
break break
if len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) == 0: if len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) == 0:
with shared.printLock: logger.debug('(concerning %s) number of '
print '(concerning', self.HOST + ')', 'number of objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave is now', len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) 'objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave '
'is now %s',
self.HOST,
len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave))
try: try:
del shared.numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer[ del shared.numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer[
@ -214,8 +219,11 @@ class receiveDataThread(threading.Thread):
except: except:
pass pass
if len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) > 0: if len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) > 0:
with shared.printLock: logger.debug('(concerning %s) number of '
print '(concerning', self.HOST + ')', 'number of objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave is now', len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) 'objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave '
'is now %s',
self.HOST,
len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave))
shared.numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer[self.HOST] = len( shared.numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer[self.HOST] = len(
self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) # this data structure is maintained so that we can keep track of how many total objects, across all connections, are currently outstanding. If it goes too high it can indicate that we are under attack by multiple nodes working together. self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) # this data structure is maintained so that we can keep track of how many total objects, across all connections, are currently outstanding. If it goes too high it can indicate that we are under attack by multiple nodes working together.
@ -234,22 +242,21 @@ class receiveDataThread(threading.Thread):
payloadLengthExtraBytes = shared.networkDefaultPayloadLengthExtraBytes payloadLengthExtraBytes = shared.networkDefaultPayloadLengthExtraBytes
POW, = unpack('>Q', hashlib.sha512(hashlib.sha512(data[ POW, = unpack('>Q', hashlib.sha512(hashlib.sha512(data[
:8] + hashlib.sha512(data[8:]).digest()).digest()).digest()[0:8]) :8] + hashlib.sha512(data[8:]).digest()).digest()).digest()[0:8])
# print 'POW:', POW # logger.debug('POW: %s', POW)
return POW <= 2 ** 64 / ((len(data) + payloadLengthExtraBytes) * (nonceTrialsPerByte)) return POW <= 2 ** 64 / ((len(data) + payloadLengthExtraBytes) * (nonceTrialsPerByte))
def sendpong(self): def sendpong(self):
print 'Sending pong' logger.debug('Sending pong')
try: try:
self.sock.sendall( self.sock.sendall(
'\xE9\xBE\xB4\xD9\x70\x6F\x6E\x67\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xcf\x83\xe1\x35') '\xE9\xBE\xB4\xD9\x70\x6F\x6E\x67\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xcf\x83\xe1\x35')
except Exception as err: except Exception as err:
# if not 'Bad file descriptor' in err: # if not 'Bad file descriptor' in err:
with shared.printLock: logger.info('sock.sendall error:', err)
print 'sock.sendall error:', err
def recverack(self): def recverack(self):
print 'verack received' logger.info('verack received')
self.verackReceived = True self.verackReceived = True
if self.verackSent: if self.verackSent:
# We have thus both sent and received a verack. # We have thus both sent and received a verack.
@ -264,18 +271,18 @@ class receiveDataThread(threading.Thread):
shared.UISignalQueue.put(('updateNetworkStatusTab', 'no data')) shared.UISignalQueue.put(('updateNetworkStatusTab', 'no data'))
remoteNodeIncomingPort, remoteNodeSeenTime = shared.knownNodes[ remoteNodeIncomingPort, remoteNodeSeenTime = shared.knownNodes[
self.streamNumber][self.HOST] self.streamNumber][self.HOST]
with shared.printLock: logger.info('Connection fully established with %s:%s. The size of the connectedHostList '
print 'Connection fully established with', self.HOST, remoteNodeIncomingPort 'is now: %s. The length of sendDataQueues is now: %s. Broadcasting addr from '
print 'The size of the connectedHostsList is now', len(shared.connectedHostsList) 'within connectionFullyEstablished().',
print 'The length of sendDataQueues is now:', len(shared.sendDataQueues) self.HOST, remoteNodeIncomingPort,
print 'broadcasting addr from within connectionFullyEstablished function.' len(shared.connectedHostsList),
len(shared.sendDataQueues))
self.broadcastaddr([(int(time.time()), self.streamNumber, 1, self.HOST, self.broadcastaddr([(int(time.time()), self.streamNumber, 1, self.HOST,
remoteNodeIncomingPort)]) # This lets all of our peers know about this new node. remoteNodeIncomingPort)]) # This lets all of our peers know about this new node.
self.sendaddr() # This is one large addr message to this one peer. self.sendaddr() # This is one large addr message to this one peer.
if not self.initiatedConnection and len(shared.connectedHostsList) > 200: if not self.initiatedConnection and len(shared.connectedHostsList) > 200:
with shared.printLock: logger.warning('We are connected to too many people. Closing connection.')
print 'We are connected to too many people. Closing connection.'
shared.broadcastToSendDataQueues((0, 'shutdown', self.HOST)) shared.broadcastToSendDataQueues((0, 'shutdown', self.HOST))
return return
@ -328,15 +335,14 @@ class receiveDataThread(threading.Thread):
headerData += 'inv\x00\x00\x00\x00\x00\x00\x00\x00\x00' headerData += 'inv\x00\x00\x00\x00\x00\x00\x00\x00\x00'
headerData += pack('>L', len(payload)) headerData += pack('>L', len(payload))
headerData += hashlib.sha512(payload).digest()[:4] headerData += hashlib.sha512(payload).digest()[:4]
with shared.printLock: logger.info('Sending huge inv message with %s objects to just this one peer',
print 'Sending huge inv message with', numberOfObjects, 'objects to just this one peer' numberOfObjects)
try: try:
self.sock.sendall(headerData + payload) self.sock.sendall(headerData + payload)
except Exception as err: except Exception as err:
# if not 'Bad file descriptor' in err: # if not 'Bad file descriptor' in err:
with shared.printLock: logger.debug('sock.sendall error: %s', err)
print 'sock.sendall error:', err
# We have received a broadcast message # We have received a broadcast message
@ -344,7 +350,7 @@ class receiveDataThread(threading.Thread):
self.messageProcessingStartTime = time.time() self.messageProcessingStartTime = time.time()
# First we must check to make sure the proof of work is sufficient. # First we must check to make sure the proof of work is sufficient.
if not self.isProofOfWorkSufficient(data): if not self.isProofOfWorkSufficient(data):
print 'Proof of work in broadcast message insufficient.' logger.info('Proof of work in broadcast message insufficient.')
return return
readPosition = 8 # bypass the nonce readPosition = 8 # bypass the nonce
embeddedTime, = unpack('>I', data[readPosition:readPosition + 4]) embeddedTime, = unpack('>I', data[readPosition:readPosition + 4])
@ -358,13 +364,16 @@ class receiveDataThread(threading.Thread):
readPosition += 4 readPosition += 4
if embeddedTime > (int(time.time()) + 10800): # prevent funny business if embeddedTime > (int(time.time()) + 10800): # prevent funny business
print 'The embedded time in this broadcast message is more than three hours in the future. That doesn\'t make sense. Ignoring message.' logger.info('The embedded time in this broadcast message is more than three hours in '
'the future. Ignoring message.')
return return
if embeddedTime < (int(time.time()) - shared.maximumAgeOfAnObjectThatIAmWillingToAccept): if embeddedTime < (int(time.time()) - shared.maximumAgeOfAnObjectThatIAmWillingToAccept):
print 'The embedded time in this broadcast message is too old. Ignoring message.' logger.info('The embedded time in this broadcast message is too old. '
'Ignoring message.')
return return
if len(data) < 180: if len(data) < 180:
print 'The payload length of this broadcast packet is unreasonably low. Someone is probably trying funny business. Ignoring message.' logger.info('The payload length of this broadcast packet is unreasonably low. '
'Someone is probably trying funny business. Ignoring message.')
return return
# Let us check to make sure the stream number is correct (thus # Let us check to make sure the stream number is correct (thus
# preventing an individual from sending broadcasts out on the wrong # preventing an individual from sending broadcasts out on the wrong
@ -375,7 +384,9 @@ class receiveDataThread(threading.Thread):
streamNumber, streamNumberLength = decodeVarint(data[ streamNumber, streamNumberLength = decodeVarint(data[
readPosition + broadcastVersionLength:readPosition + broadcastVersionLength + 10]) readPosition + broadcastVersionLength:readPosition + broadcastVersionLength + 10])
if streamNumber != self.streamNumber: if streamNumber != self.streamNumber:
print 'The stream number encoded in this broadcast message (' + str(streamNumber) + ') does not match the stream number on which it was received. Ignoring it.' logger.debug('The stream number encoded in this broadcast message (%s) does not '
'match the stream number on which it was received. Ignoring it.',
streamNumber)
return return
shared.inventoryLock.acquire() shared.inventoryLock.acquire()
@ -1598,228 +1609,87 @@ class receiveDataThread(threading.Thread):
# We have received an addr message. # We have received an addr message.
def recaddr(self, data): def recaddr(self, data):
listOfAddressDetailsToBroadcastToPeers = [] listOfAddressDetailsToBroadcastToPeers = []
numberOfAddressesIncluded = 0
numberOfAddressesIncluded, lengthOfNumberOfAddresses = decodeVarint(
data[:10])
if shared.verbose >= 1: from message_parsers import AddressMessageParser
with shared.printLock: addrParser = AddressMessageParser(data, self.remoteProtocolVersion)
print 'addr message contains', numberOfAddressesIncluded, 'IP addresses.' needToWriteKnownNodesToDisk = False
for hostDetails in addrParser.parse():
rec_timestamp, rec_stream, rec_services, rec_host, rec_port = hostDetails
if self.remoteProtocolVersion == 1: if not rec_host.startswith('::ffff:'):
if numberOfAddressesIncluded > 1000 or numberOfAddressesIncluded == 0: logger.debug('Skipping IPv6 address %s.' % (rec_host))
return continue
if len(data) != lengthOfNumberOfAddresses + (34 * numberOfAddressesIncluded):
print 'addr message does not contain the correct amount of data. Ignoring.'
return
needToWriteKnownNodesToDisk = False # Trim off leading ::ffff: to make IPv4-compatible.
for i in range(0, numberOfAddressesIncluded): rec_host = rec_host[7:]
try:
if data[16 + lengthOfNumberOfAddresses + (34 * i):28 + lengthOfNumberOfAddresses + (34 * i)] != '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF':
with shared.printLock:
print 'Skipping IPv6 address.', repr(data[16 + lengthOfNumberOfAddresses + (34 * i):28 + lengthOfNumberOfAddresses + (34 * i)])
continue if helper_generic.isHostInPrivateIPRange(rec_host):
except Exception as err: logger.debug('Skipping IP in private range: %s' % (rec_host))
with shared.printLock: continue
sys.stderr.write( if helper_generic.isHostInLoopbackIPRange(rec_host):
'ERROR TRYING TO UNPACK recaddr (to test for an IPv6 address). Message: %s\n' % str(err)) logger.debug('Skipping IP in loopback range: %s' % (rec_host))
continue
break # giving up on unpacking any more. We should still be connected however. # knownNodes is a dictionary of dictionaries with one outer dictionary for each stream.
# If the outer stream dictionary doesn't exist yet then we must make it.
try: if rec_stream not in shared.knownNodes:
recaddrStream, = unpack('>I', data[4 + lengthOfNumberOfAddresses + (
34 * i):8 + lengthOfNumberOfAddresses + (34 * i)])
except Exception as err:
with shared.printLock:
sys.stderr.write(
'ERROR TRYING TO UNPACK recaddr (recaddrStream). Message: %s\n' % str(err))
break # giving up on unpacking any more. We should still be connected however.
if recaddrStream == 0:
continue
if recaddrStream != self.streamNumber and recaddrStream != (self.streamNumber * 2) and recaddrStream != ((self.streamNumber * 2) + 1): # if the embedded stream number is not in my stream or either of my child streams then ignore it. Someone might be trying funny business.
continue
try:
recaddrServices, = unpack('>Q', data[8 + lengthOfNumberOfAddresses + (
34 * i):16 + lengthOfNumberOfAddresses + (34 * i)])
except Exception as err:
with shared.printLock:
sys.stderr.write(
'ERROR TRYING TO UNPACK recaddr (recaddrServices). Message: %s\n' % str(err))
break # giving up on unpacking any more. We should still be connected however.
try:
recaddrPort, = unpack('>H', data[32 + lengthOfNumberOfAddresses + (
34 * i):34 + lengthOfNumberOfAddresses + (34 * i)])
except Exception as err:
with shared.printLock:
sys.stderr.write(
'ERROR TRYING TO UNPACK recaddr (recaddrPort). Message: %s\n' % str(err))
break # giving up on unpacking any more. We should still be connected however.
# print 'Within recaddr(): IP', recaddrIP, ', Port',
# recaddrPort, ', i', i
hostFromAddrMessage = socket.inet_ntoa(data[
28 + lengthOfNumberOfAddresses + (34 * i):32 + lengthOfNumberOfAddresses + (34 * i)])
# print 'hostFromAddrMessage', hostFromAddrMessage
if data[28 + lengthOfNumberOfAddresses + (34 * i)] == '\x7F':
print 'Ignoring IP address in loopback range:', hostFromAddrMessage
continue
if helper_generic.isHostInPrivateIPRange(hostFromAddrMessage):
print 'Ignoring IP address in private range:', hostFromAddrMessage
continue
timeSomeoneElseReceivedMessageFromThisNode, = unpack('>I', data[lengthOfNumberOfAddresses + (
34 * i):4 + lengthOfNumberOfAddresses + (34 * i)]) # This is the 'time' value in the received addr message.
if recaddrStream not in shared.knownNodes: # knownNodes is a dictionary of dictionaries with one outer dictionary for each stream. If the outer stream dictionary doesn't exist yet then we must make it.
shared.knownNodesLock.acquire()
shared.knownNodes[recaddrStream] = {}
shared.knownNodesLock.release()
if hostFromAddrMessage not in shared.knownNodes[recaddrStream]:
if len(shared.knownNodes[recaddrStream]) < 20000 and timeSomeoneElseReceivedMessageFromThisNode > (int(time.time()) - 10800) and timeSomeoneElseReceivedMessageFromThisNode < (int(time.time()) + 10800): # If we have more than 20000 nodes in our list already then just forget about adding more. Also, make sure that the time that someone else received a message from this node is within three hours from now.
shared.knownNodesLock.acquire()
shared.knownNodes[recaddrStream][hostFromAddrMessage] = (
recaddrPort, timeSomeoneElseReceivedMessageFromThisNode)
shared.knownNodesLock.release()
needToWriteKnownNodesToDisk = True
hostDetails = (
timeSomeoneElseReceivedMessageFromThisNode,
recaddrStream, recaddrServices, hostFromAddrMessage, recaddrPort)
listOfAddressDetailsToBroadcastToPeers.append(
hostDetails)
else:
PORT, timeLastReceivedMessageFromThisNode = shared.knownNodes[recaddrStream][
hostFromAddrMessage] # PORT in this case is either the port we used to connect to the remote node, or the port that was specified by someone else in a past addr message.
if (timeLastReceivedMessageFromThisNode < timeSomeoneElseReceivedMessageFromThisNode) and (timeSomeoneElseReceivedMessageFromThisNode < int(time.time())):
shared.knownNodesLock.acquire()
shared.knownNodes[recaddrStream][hostFromAddrMessage] = (
PORT, timeSomeoneElseReceivedMessageFromThisNode)
shared.knownNodesLock.release()
if PORT != recaddrPort:
print 'Strange occurance: The port specified in an addr message', str(recaddrPort), 'does not match the port', str(PORT), 'that this program (or some other peer) used to connect to it', str(hostFromAddrMessage), '. Perhaps they changed their port or are using a strange NAT configuration.'
if needToWriteKnownNodesToDisk: # Runs if any nodes were new to us. Also, share those nodes with our peers.
shared.knownNodesLock.acquire() shared.knownNodesLock.acquire()
output = open(shared.appdata + 'knownnodes.dat', 'wb') shared.knownNodes[rec_stream] = {}
pickle.dump(shared.knownNodes, output)
output.close()
shared.knownNodesLock.release() shared.knownNodesLock.release()
self.broadcastaddr(
listOfAddressDetailsToBroadcastToPeers) # no longer broadcast
with shared.printLock:
print 'knownNodes currently has', len(shared.knownNodes[self.streamNumber]), 'nodes for this stream.'
elif self.remoteProtocolVersion >= 2: # The difference is that in protocol version 2, network addresses use 64 bit times rather than 32 bit times. if rec_host not in shared.knownNodes[rec_stream]:
if numberOfAddressesIncluded > 1000 or numberOfAddressesIncluded == 0: # If we have more than 20000 nodes in our list already then just forget about adding
return # more. Also, make sure that the time that someone else received a message from this
if len(data) != lengthOfNumberOfAddresses + (38 * numberOfAddressesIncluded): # node is within three hours from now.
print 'addr message does not contain the correct amount of data. Ignoring.' if len(shared.knownNodes[rec_stream]) < 20000 \
return and rec_timestamp > (int(time.time()) - 10800) \
and rec_timestamp < (int(time.time()) + 10800):
needToWriteKnownNodesToDisk = False
for i in range(0, numberOfAddressesIncluded):
try:
if data[20 + lengthOfNumberOfAddresses + (38 * i):32 + lengthOfNumberOfAddresses + (38 * i)] != '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF':
with shared.printLock:
print 'Skipping IPv6 address.', repr(data[20 + lengthOfNumberOfAddresses + (38 * i):32 + lengthOfNumberOfAddresses + (38 * i)])
continue
except Exception as err:
with shared.printLock:
sys.stderr.write(
'ERROR TRYING TO UNPACK recaddr (to test for an IPv6 address). Message: %s\n' % str(err))
break # giving up on unpacking any more. We should still be connected however.
try:
recaddrStream, = unpack('>I', data[8 + lengthOfNumberOfAddresses + (
38 * i):12 + lengthOfNumberOfAddresses + (38 * i)])
except Exception as err:
with shared.printLock:
sys.stderr.write(
'ERROR TRYING TO UNPACK recaddr (recaddrStream). Message: %s\n' % str(err))
break # giving up on unpacking any more. We should still be connected however.
if recaddrStream == 0:
continue
if recaddrStream != self.streamNumber and recaddrStream != (self.streamNumber * 2) and recaddrStream != ((self.streamNumber * 2) + 1): # if the embedded stream number is not in my stream or either of my child streams then ignore it. Someone might be trying funny business.
continue
try:
recaddrServices, = unpack('>Q', data[12 + lengthOfNumberOfAddresses + (
38 * i):20 + lengthOfNumberOfAddresses + (38 * i)])
except Exception as err:
with shared.printLock:
sys.stderr.write(
'ERROR TRYING TO UNPACK recaddr (recaddrServices). Message: %s\n' % str(err))
break # giving up on unpacking any more. We should still be connected however.
try:
recaddrPort, = unpack('>H', data[36 + lengthOfNumberOfAddresses + (
38 * i):38 + lengthOfNumberOfAddresses + (38 * i)])
except Exception as err:
with shared.printLock:
sys.stderr.write(
'ERROR TRYING TO UNPACK recaddr (recaddrPort). Message: %s\n' % str(err))
break # giving up on unpacking any more. We should still be connected however.
# print 'Within recaddr(): IP', recaddrIP, ', Port',
# recaddrPort, ', i', i
hostFromAddrMessage = socket.inet_ntoa(data[
32 + lengthOfNumberOfAddresses + (38 * i):36 + lengthOfNumberOfAddresses + (38 * i)])
# print 'hostFromAddrMessage', hostFromAddrMessage
if data[32 + lengthOfNumberOfAddresses + (38 * i)] == '\x7F':
print 'Ignoring IP address in loopback range:', hostFromAddrMessage
continue
if data[32 + lengthOfNumberOfAddresses + (38 * i)] == '\x0A':
print 'Ignoring IP address in private range:', hostFromAddrMessage
continue
if data[32 + lengthOfNumberOfAddresses + (38 * i):34 + lengthOfNumberOfAddresses + (38 * i)] == '\xC0A8':
print 'Ignoring IP address in private range:', hostFromAddrMessage
continue
timeSomeoneElseReceivedMessageFromThisNode, = unpack('>Q', data[lengthOfNumberOfAddresses + (
38 * i):8 + lengthOfNumberOfAddresses + (38 * i)]) # This is the 'time' value in the received addr message. 64-bit.
if recaddrStream not in shared.knownNodes: # knownNodes is a dictionary of dictionaries with one outer dictionary for each stream. If the outer stream dictionary doesn't exist yet then we must make it.
shared.knownNodesLock.acquire() shared.knownNodesLock.acquire()
shared.knownNodes[recaddrStream] = {} shared.knownNodes[rec_stream][rec_host] = (
rec_port, rec_timestamp)
shared.knownNodesLock.release() shared.knownNodesLock.release()
if hostFromAddrMessage not in shared.knownNodes[recaddrStream]:
if len(shared.knownNodes[recaddrStream]) < 20000 and timeSomeoneElseReceivedMessageFromThisNode > (int(time.time()) - 10800) and timeSomeoneElseReceivedMessageFromThisNode < (int(time.time()) + 10800): # If we have more than 20000 nodes in our list already then just forget about adding more. Also, make sure that the time that someone else received a message from this node is within three hours from now.
shared.knownNodesLock.acquire()
shared.knownNodes[recaddrStream][hostFromAddrMessage] = (
recaddrPort, timeSomeoneElseReceivedMessageFromThisNode)
shared.knownNodesLock.release()
with shared.printLock:
print 'added new node', hostFromAddrMessage, 'to knownNodes in stream', recaddrStream
needToWriteKnownNodesToDisk = True needToWriteKnownNodesToDisk = True
hostDetails = (
timeSomeoneElseReceivedMessageFromThisNode, listOfAddressDetailsToBroadcastToPeers.append(hostDetails)
recaddrStream, recaddrServices, hostFromAddrMessage, recaddrPort)
listOfAddressDetailsToBroadcastToPeers.append( else:
hostDetails) # Just makes things a little more readable below.
else: timeSomeoneElseReceivedMessageFromThisNode = rec_timestamp
PORT, timeLastReceivedMessageFromThisNode = shared.knownNodes[recaddrStream][
hostFromAddrMessage] # PORT in this case is either the port we used to connect to the remote node, or the port that was specified by someone else in a past addr message. # PORT in this case is either the port we used to connect to the remote node, or
if (timeLastReceivedMessageFromThisNode < timeSomeoneElseReceivedMessageFromThisNode) and (timeSomeoneElseReceivedMessageFromThisNode < int(time.time())): # the port that was specified by someone else in a past addr message.
shared.knownNodesLock.acquire() PORT, timeLastReceivedMessageFromThisNode = \
shared.knownNodes[recaddrStream][hostFromAddrMessage] = ( shared.knownNodes[rec_stream][rec_host]
PORT, timeSomeoneElseReceivedMessageFromThisNode) if (timeLastReceivedMessageFromThisNode < timeSomeoneElseReceivedMessageFromThisNode) \
shared.knownNodesLock.release() and (timeSomeoneElseReceivedMessageFromThisNode < int(time.time())):
if PORT != recaddrPort: shared.knownNodesLock.acquire()
print 'Strange occurance: The port specified in an addr message', str(recaddrPort), 'does not match the port', str(PORT), 'that this program (or some other peer) used to connect to it', str(hostFromAddrMessage), '. Perhaps they changed their port or are using a strange NAT configuration.' shared.knownNodes[rec_stream][rec_host] = (
if needToWriteKnownNodesToDisk: # Runs if any nodes were new to us. Also, share those nodes with our peers. PORT, timeSomeoneElseReceivedMessageFromThisNode)
shared.knownNodesLock.acquire() shared.knownNodesLock.release()
output = open(shared.appdata + 'knownnodes.dat', 'wb') if PORT != rec_port:
pickle.dump(shared.knownNodes, output) logger.debug('Strange occurance: The port specified in an addr message '
output.close() '%s does not match the port %s that this program (or some '
shared.knownNodesLock.release() 'other peer) used to connect to the host %s. Perhaps they '
self.broadcastaddr(listOfAddressDetailsToBroadcastToPeers) 'changed their port of are using a strange NAT configuration.'
with shared.printLock: % (rec_port, PORT, rec_host))
print 'knownNodes currently has', len(shared.knownNodes[self.streamNumber]), 'nodes for this stream.'
# Runs if any nodes were new to us. Also, share those nodes with our peers.
if needToWriteKnownNodesToDisk:
shared.knownNodesLock.acquire()
output = open(shared.appdata + 'knownnodes.dat', 'wb')
pickle.dump(shared.knownNodes, output)
output.close()
shared.knownNodesLock.release()
self.broadcastaddr(
listOfAddressDetailsToBroadcastToPeers) # no longer broadcast
logger.info('knownNodes currently has %s nodes for this stream.'
% (len(shared.knownNodes[self.streamNumber])))
# Function runs when we want to broadcast an addr message to all of our # Function runs when we want to broadcast an addr message to all of our
@ -1836,8 +1706,7 @@ class receiveDataThread(threading.Thread):
payload += pack('>I', streamNumber) payload += pack('>I', streamNumber)
payload += pack( payload += pack(
'>q', services) # service bit flags offered by this node '>q', services) # service bit flags offered by this node
payload += '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF' + \ payload += shared.packNetworkAddress(host)
socket.inet_aton(host)
payload += pack('>H', port) # remote port payload += pack('>H', port) # remote port
payload = encodeVarint(numberOfAddressesInAddrMessage) + payload payload = encodeVarint(numberOfAddressesInAddrMessage) + payload
@ -1903,8 +1772,7 @@ class receiveDataThread(threading.Thread):
payload += pack('>I', self.streamNumber) payload += pack('>I', self.streamNumber)
payload += pack( payload += pack(
'>q', 1) # service bit flags offered by this node '>q', 1) # service bit flags offered by this node
payload += '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF' + \ payload += shared.packNetworkAddress(HOST)
socket.inet_aton(HOST)
payload += pack('>H', PORT) # remote port payload += pack('>H', PORT) # remote port
for HOST, value in addrsInChildStreamLeft.items(): for HOST, value in addrsInChildStreamLeft.items():
PORT, timeLastReceivedMessageFromThisNode = value PORT, timeLastReceivedMessageFromThisNode = value
@ -1915,8 +1783,7 @@ class receiveDataThread(threading.Thread):
payload += pack('>I', self.streamNumber * 2) payload += pack('>I', self.streamNumber * 2)
payload += pack( payload += pack(
'>q', 1) # service bit flags offered by this node '>q', 1) # service bit flags offered by this node
payload += '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF' + \ payload += packNetworkAddress(HOST)
socket.inet_aton(HOST)
payload += pack('>H', PORT) # remote port payload += pack('>H', PORT) # remote port
for HOST, value in addrsInChildStreamRight.items(): for HOST, value in addrsInChildStreamRight.items():
PORT, timeLastReceivedMessageFromThisNode = value PORT, timeLastReceivedMessageFromThisNode = value
@ -1927,8 +1794,7 @@ class receiveDataThread(threading.Thread):
payload += pack('>I', (self.streamNumber * 2) + 1) payload += pack('>I', (self.streamNumber * 2) + 1)
payload += pack( payload += pack(
'>q', 1) # service bit flags offered by this node '>q', 1) # service bit flags offered by this node
payload += '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF' + \ payload += shared.packNetworkAddress(HOST)
socket.inet_aton(HOST)
payload += pack('>H', PORT) # remote port payload += pack('>H', PORT) # remote port
payload = encodeVarint(numberOfAddressesInAddrMessage) + payload payload = encodeVarint(numberOfAddressesInAddrMessage) + payload

6
src/helper_generic.py
View File

@ -32,3 +32,9 @@ def isHostInPrivateIPRange(host):
if host[:8] == '192.168.': if host[:8] == '192.168.':
return True return True
return False return False
def isHostInLoopbackIPRange(host):
if host[:4] == '127.':
return True
else:
return False

125
src/message_parsers.py Normal file
View File

@ -0,0 +1,125 @@
from struct import unpack
from addresses import decodeVarint
from debug import logger
from shared import unpackNetworkAddress
class AddressMessageParser:
def __init__(self, data, remoteProtocolVersion):
# Byte array of data to parse.
self.data = data
# Position of where the next datum starts, in bytes.
self.position = 0
# Protocol version of the peer with which this message corresponds.
self.remoteProtocolVersion = remoteProtocolVersion
def parse(self):
#logger.debug('Parsing addr message %s' % (repr(self.data)))
listOfAddressDetailsToBroadcast = []
numberOfAddresses = self.__consumeNumberOfAddresses()
if not numberOfAddresses:
raise StopIteration
logger.debug('addr message contains %d IP addresses.', numberOfAddresses)
needToWriteKnownNodes = False
for addressIndex in range(numberOfAddresses):
try:
hostDetails = self.__consumeAddress()
except Exception as err:
logger.exception('ERROR TRYING TO UNPACK addr message.')
break
if not hostDetails:
continue
break
timestamp, stream, services, host, port = hostDetails
yield hostDetails
def __consumeNumberOfAddresses(self):
numberOfAddresses, lengthOfNumberOfAddresses = decodeVarint(
self.data[self.position:self.position+10])
self.position += lengthOfNumberOfAddresses
# Sanity checks
if numberOfAddresses > 1000:
logger.debug('addr message contains too many addresses. Ignoring.')
return 0
if numberOfAddresses == 0:
logger.debug('addr message contains no addresses.')
return 0
if ( self.remoteProtocolVersion == 1 and
len(self.data) != lengthOfNumberOfAddresses + (34 * numberOfAddresses)
) or (
self.remoteProtocolVersion == 2 and
len(self.data) != lengthOfNumberOfAddresses + (38 * numberOfAddresses)
):
logger.debug('addr message (%s) does not contain the correct amount of data. Ignoring.'
% (repr(self.data)))
return 0
return numberOfAddresses
def __consumeTimestamp(self):
if self.remoteProtocolVersion == 1:
#logging.debug('at %s, timestamp data: %s'
# % (self.position, repr(self.data[self.position : self.position + 4])))
timestamp, = unpack(
'>I', self.data[self.position : self.position + 4])
self.position += 4
elif self.remoteProtocolVersion == 2:
#logger.debug('at %s, timestamp data: %s'
# % (self.position, repr(self.data[self.position : self.position + 8])))
timestamp, = unpack(
'>Q', self.data[self.position : self.position + 8])
self.position += 8
return timestamp
def __consumeStream(self):
#logger.debug('at %s, stream data: %s'
# % (self.position, repr(self.data[self.position : self.position + 4])))
stream, = unpack('>I', self.data[self.position : self.position + 4])
self.position += 4
return stream
def __consumeServices(self):
#logger.debug('at %s, services data: %s'
# % (self.position, repr(self.data[self.position : self.position + 8])))
services, = unpack('>Q', self.data[self.position : self.position + 8])
self.position += 8
return services
def __consumeHost(self):
#logger.debug('at %s, host data: %s'
# % (self.position, repr(self.data[self.position : self.position + 16])))
hostdata = self.data[self.position : self.position + 16]
self.position += 16
host = unpackNetworkAddress(hostdata)
#logger.debug('host: %s' % (host))
return host
def __consumePort(self):
#logger.debug('at %s, port data: %s'
# % (self.position, repr(self.data[self.position : self.position + 2])))
port, = unpack('>H', self.data[self.position : self.position + 2])
#logger.debug('port: %s' % (port))
self.position += 2
return port
# Consume one entry in the addr_list.
# Returns (timestamp, stream, services, host, port) tuple on success.
def __consumeAddress(self):
past_position = self.position
try:
timestamp = self.__consumeTimestamp()
stream = self.__consumeStream()
services = self.__consumeServices()
host = self.__consumeHost()
port = self.__consumePort()
finally:
if self.remoteProtocolVersion == 1:
self.position = past_position + 34
elif self.remoteProtocolVersion == 2:
self.position = past_position + 38
return (timestamp, stream, services, host, port)

24
src/shared.py
View File

@ -77,6 +77,26 @@ def isInSqlInventory(hash):
else: else:
return True return True
def packNetworkAddress(address):
try:
# Matches IPV4-style address.
if ':' not in address and address.count('.') == 3:
return socket.inet_pton(socket.AF_INET6, '::ffff:' + address)
# Matches IPV4-mapped IPV6 and plain IPV6.
else:
return socket.inet_pton(socket.AF_INET6, address)
except OSError:
logger.error('Failed to pack address "%s".' % (address))
raise
def unpackNetworkAddress(packedAddress):
try:
address = socket.inet_ntop(socket.AF_INET6, packedAddress)
return address
except:
logger.error('Failed to unpack address %s.' % repr(packedAddress))
raise
def assembleVersionMessage(remoteHost, remotePort, myStreamNumber): def assembleVersionMessage(remoteHost, remotePort, myStreamNumber):
payload = '' payload = ''
payload += pack('>L', 2) # protocol version. payload += pack('>L', 2) # protocol version.
@ -85,8 +105,8 @@ def assembleVersionMessage(remoteHost, remotePort, myStreamNumber):
payload += pack( payload += pack(
'>q', 1) # boolservices of remote connection. How can I even know this for sure? This is probably ignored by the remote host. '>q', 1) # boolservices of remote connection. How can I even know this for sure? This is probably ignored by the remote host.
payload += '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xFF\xFF' + \
socket.inet_aton(remoteHost) payload += packNetworkAddress(remoteHost)
payload += pack('>H', remotePort) # remote IPv6 and port payload += pack('>H', remotePort) # remote IPv6 and port
payload += pack('>q', 1) # bitflags of the services I offer. payload += pack('>q', 1) # bitflags of the services I offer.