From 87cbfe9434d399ea54f7a9f0262d02c897f982a1 Mon Sep 17 00:00:00 2001 From: Gregor Robinson Date: Thu, 20 Jun 2013 13:17:11 -0600 Subject: [PATCH 1/6] Continuing migration from print-based logging. --- src/bitmessagemain.py | 182 +++++++++++++++++++----------------------- 1 file changed, 83 insertions(+), 99 deletions(-) diff --git a/src/bitmessagemain.py b/src/bitmessagemain.py index 0b93cdcf..b4cf7da7 100644 --- a/src/bitmessagemain.py +++ b/src/bitmessagemain.py @@ -8,7 +8,7 @@ # yet contain logic to expand into further streams. # The software version variable is now held in shared.py -verbose = 1 +verbose=2 maximumAgeOfAnObjectThatIAmWillingToAccept = 216000 # Equals two days and 12 hours. lengthOfTimeToLeaveObjectsInInventory = 237600 # Equals two days and 18 hours. This should be longer than maximumAgeOfAnObjectThatIAmWillingToAccept so that we don't process messages twice. lengthOfTimeToHoldOnToAllPubkeys = 2419200 # Equals 4 weeks. You could make this longer if you want but making it shorter would not be advisable because there is a very small possibility that it could keep you from obtaining a needed pubkey for a period of time. @@ -48,6 +48,7 @@ from subprocess import call # used when the API must execute an outside program import singleton import proofofwork # Logging +# TODO(fiatflux): Add option to suppress extra logging info (e.g. tracebacks) in stdout/stderr. from debug import logger # For each stream to which we connect, several outgoingSynSender threads @@ -65,7 +66,8 @@ class outgoingSynSender(threading.Thread): time.sleep(1) global alreadyAttemptedConnectionsListResetTime while True: - while len(selfInitiatedConnections[self.streamNumber]) >= 8: # maximum number of outgoing connections = 8 + # Maximum number of outgoing connections = 8. + while len(selfInitiatedConnections[self.streamNumber]) >= 8: time.sleep(10) if shared.shutdown: break @@ -74,7 +76,7 @@ class outgoingSynSender(threading.Thread): alreadyAttemptedConnectionsListLock.acquire() while HOST in alreadyAttemptedConnectionsList or HOST in shared.connectedHostsList: alreadyAttemptedConnectionsListLock.release() - # print 'choosing new sample' + logger.debug('Choosing new sample.') random.seed() HOST, = random.sample(shared.knownNodes[ self.streamNumber], 1) @@ -96,23 +98,18 @@ class outgoingSynSender(threading.Thread): # can rebind faster sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.settimeout(20) - if shared.config.get('bitmessagesettings', 'socksproxytype') == 'none' and verbose >= 2: - #shared.printLock.acquire() - #print 'Trying an outgoing connection to', HOST, ':', PORT - logger.info('Trying an outgoing connection to %s:%s' % (HOST, PORT)) - #shared.printLock.release() - # sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + if shared.config.get('bitmessagesettings', 'socksproxytype') == 'none': + logger.debug('Trying an outgoing connection to %s:%s' % (HOST, PORT)) elif shared.config.get('bitmessagesettings', 'socksproxytype') == 'SOCKS4a': - if verbose >= 2: - shared.printLock.acquire() - print '(Using SOCKS4a) Trying an outgoing connection to', HOST, ':', PORT - shared.printLock.release() + logger.debug('(Using SOCKS4a) Trying an outgoing connection to %s:%s' % (HOST, PORT)) proxytype = socks.PROXY_TYPE_SOCKS4 sockshostname = shared.config.get( 'bitmessagesettings', 'sockshostname') socksport = shared.config.getint( 'bitmessagesettings', 'socksport') - rdns = True # Do domain name lookups through the proxy; though this setting doesn't really matter since we won't be doing any domain name lookups anyway. + # Do domain name lookups through the proxy; though this setting doesn't really matter since we won't be + # doing any domain name lookups anyway. + rdns = True if shared.config.getboolean('bitmessagesettings', 'socksauthentication'): socksusername = shared.config.get( 'bitmessagesettings', 'socksusername') @@ -124,16 +121,15 @@ class outgoingSynSender(threading.Thread): sock.setproxy( proxytype, sockshostname, socksport, rdns) elif shared.config.get('bitmessagesettings', 'socksproxytype') == 'SOCKS5': - if verbose >= 2: - shared.printLock.acquire() - print '(Using SOCKS5) Trying an outgoing connection to', HOST, ':', PORT - shared.printLock.release() + logger.debug('(Using SOCKS5) Trying an outgoing connection to %s:%s' % (HOST, PORT)) proxytype = socks.PROXY_TYPE_SOCKS5 sockshostname = shared.config.get( 'bitmessagesettings', 'sockshostname') socksport = shared.config.getint( 'bitmessagesettings', 'socksport') - rdns = True # Do domain name lookups through the proxy; though this setting doesn't really matter since we won't be doing any domain name lookups anyway. + # Do domain name lookups through the proxy; though this setting doesn't really matter since we won't be + # doing any domain name lookups anyway. + rdns = True if shared.config.getboolean('bitmessagesettings', 'socksauthentication'): socksusername = shared.config.get( 'bitmessagesettings', 'socksusername') @@ -153,9 +149,7 @@ class outgoingSynSender(threading.Thread): rd.setup(sock, HOST, PORT, self.streamNumber, objectsOfWhichThisRemoteNodeIsAlreadyAware) rd.start() - shared.printLock.acquire() - print self, 'connected to', HOST, 'during an outgoing attempt.' - shared.printLock.release() + logger.info('%s connected to %s during an outgoing attempt.' % (self, HOST)) sd = sendDataThread() sd.setup(sock, HOST, PORT, self.streamNumber, @@ -164,48 +158,43 @@ class outgoingSynSender(threading.Thread): sd.sendVersionMessage() except socks.GeneralProxyError as err: - if verbose >= 2: - shared.printLock.acquire() - print 'Could NOT connect to', HOST, 'during outgoing attempt.', err - shared.printLock.release() + # TODO(fiatflux): turn off traceback, but preserve terse exception info. + logger.debug('Could NOT connect to %s during outgoing attempt.', HOST, exc_info=True) PORT, timeLastSeen = shared.knownNodes[ self.streamNumber][HOST] - if (int(time.time()) - timeLastSeen) > 172800 and len(shared.knownNodes[self.streamNumber]) > 1000: # for nodes older than 48 hours old if we have more than 1000 hosts in our list, delete from the shared.knownNodes data-structure. + # for nodes older than 48 hours old if we have more than 1000 hosts in our list, delete from the + # shared.knownNodes data-structure. + if (int(time.time()) - timeLastSeen) > 172800 and len(shared.knownNodes[self.streamNumber]) > 1000: shared.knownNodesLock.acquire() del shared.knownNodes[self.streamNumber][HOST] shared.knownNodesLock.release() - shared.printLock.acquire() - print 'deleting ', HOST, 'from shared.knownNodes because it is more than 48 hours old and we could not connect to it.' - shared.printLock.release() + logger.debug('deleting %s from shared.knownNodes because it is more than 48 hours old and we ' + 'could not connect to it.' % (HOST)) except socks.Socks5AuthError as err: shared.UISignalQueue.put(( 'updateStatusBar', _translate( "MainWindow", "SOCKS5 Authentication problem: %1").arg(str(err)))) except socks.Socks5Error as err: - pass - print 'SOCKS5 error. (It is possible that the server wants authentication).)', str(err) + logger.exception('SOCKS5 error. It is possible that the server wants authentication.') except socks.Socks4Error as err: - print 'Socks4Error:', err + logger.exception('Socks4Error') except socket.error as err: if shared.config.get('bitmessagesettings', 'socksproxytype')[0:5] == 'SOCKS': - print 'Bitmessage MIGHT be having trouble connecting to the SOCKS server. ' + str(err) + logger.exception('Bitmessage MIGHT be having trouble connecting to the SOCKS server.') else: - if verbose >= 1: - shared.printLock.acquire() - print 'Could NOT connect to', HOST, 'during outgoing attempt.', err - shared.printLock.release() + logger.info('Could NOT connect to %s during outgoing attempt.', HOST, exc_info=True) PORT, timeLastSeen = shared.knownNodes[ self.streamNumber][HOST] - if (int(time.time()) - timeLastSeen) > 172800 and len(shared.knownNodes[self.streamNumber]) > 1000: # for nodes older than 48 hours old if we have more than 1000 hosts in our list, delete from the knownNodes data-structure. + # for nodes older than 48 hours old if we have more than 1000 hosts in our list, delete from the + # knownNodes data-structure. + if (int(time.time()) - timeLastSeen) > 172800 and len(shared.knownNodes[self.streamNumber]) > 1000: shared.knownNodesLock.acquire() del shared.knownNodes[self.streamNumber][HOST] shared.knownNodesLock.release() - shared.printLock.acquire() - print 'deleting ', HOST, 'from knownNodes because it is more than 48 hours old and we could not connect to it.' - shared.printLock.release() + logger.info('Deleting %s from knownNodes because it is more than 48 hours old and we ' + 'could not connect to it.' % (HOST)) except Exception as err: - sys.stderr.write( - 'An exception has occurred in the outgoingSynSender thread that was not caught by other exception types: %s\n' % err) + logger.exception('An uncaught exception has occurred in the outgoingSynSender thread.') time.sleep(0.1) # Only one singleListener thread will ever exist. It creates the @@ -228,9 +217,7 @@ class singleListener(threading.Thread): while shared.config.get('bitmessagesettings', 'socksproxytype')[0:5] == 'SOCKS': time.sleep(300) - shared.printLock.acquire() - print 'Listening for incoming connections.' - shared.printLock.release() + logger.info('Listening for incoming connections.') HOST = '' # Symbolic name meaning all available interfaces PORT = shared.config.getint('bitmessagesettings', 'port') sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -247,9 +234,7 @@ class singleListener(threading.Thread): while shared.config.get('bitmessagesettings', 'socksproxytype')[0:5] == 'SOCKS': time.sleep(10) while len(shared.connectedHostsList) > 220: - shared.printLock.acquire() - print 'We are connected to too many people. Not accepting further incoming connections for ten seconds.' - shared.printLock.release() + logger.info('We are connected to too many people. Not accepting further incoming connections for ten seconds.') time.sleep(10) a, (HOST, PORT) = sock.accept() @@ -258,9 +243,7 @@ class singleListener(threading.Thread): # because the two computers will share the same external IP. This # is here to prevent connection flooding. while HOST in shared.connectedHostsList: - shared.printLock.acquire() - print 'We are already connected to', HOST + '. Ignoring connection.' - shared.printLock.release() + logger.warning('We are already connected to', HOST + '. Ignoring connection.') a.close() a, (HOST, PORT) = sock.accept() objectsOfWhichThisRemoteNodeIsAlreadyAware = {} @@ -277,9 +260,7 @@ class singleListener(threading.Thread): a, HOST, PORT, -1, objectsOfWhichThisRemoteNodeIsAlreadyAware) rd.start() - shared.printLock.acquire() - print self, 'connected to', HOST, 'during INCOMING request.' - shared.printLock.release() + logger.info(self, 'connected to %s during INCOMING request.' % (HOST)) # This thread is created either by the synSenderThread(for outgoing # connections) or the singleListenerThread(for incoming connectiosn). @@ -304,106 +285,109 @@ class receiveDataThread(threading.Thread): self.HOST = HOST self.PORT = port self.streamNumber = streamNumber - self.payloadLength = 0 # This is the protocol payload length thus it doesn't include the 24 byte message header + # This is the protocol payload length thus it does NOT include the 24 byte message header. + self.payloadLength = 0 self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave = {} + + # The very fact that this receiveData thread exists shows that we are connected to the remote host. Let's add it + # to this list so that an outgoingSynSender thread doesn't try to connect to it.. shared.connectedHostsList[ - self.HOST] = 0 # The very fact that this receiveData thread exists shows that we are connected to the remote host. Let's add it to this list so that an outgoingSynSender thread doesn't try to connect to it. - self.connectionIsOrWasFullyEstablished = False # set to true after the remote node and I accept each other's version messages. This is needed to allow the user interface to accurately reflect the current number of connections. - if self.streamNumber == -1: # This was an incoming connection. Send out a version message if we accept the other node's version message. + self.HOST] = 0 + + # set to true after the remote node and I accept each other's version messages. This is needed to allow the user + # interface to accurately reflect the current number of connections. + self.connectionIsOrWasFullyEstablished = False + + # This was an incoming connection. Send out a version message if we accept the other node's version message. + if self.streamNumber == -1: self.initiatedConnection = False else: self.initiatedConnection = True selfInitiatedConnections[streamNumber][self] = 0 + + # When we receive a message bound for us, we store the acknowledgement that we need to send (the ackdata) here + # until we are done processing all other data received from this peer. self.ackDataThatWeHaveYetToSend = [ - ] # When we receive a message bound for us, we store the acknowledgement that we need to send (the ackdata) here until we are done processing all other data received from this peer. + ] self.objectsOfWhichThisRemoteNodeIsAlreadyAware = objectsOfWhichThisRemoteNodeIsAlreadyAware def run(self): - shared.printLock.acquire() - print 'ID of the receiveDataThread is', str(id(self)) + '. The size of the shared.connectedHostsList is now', len(shared.connectedHostsList) - shared.printLock.release() + logger.info( + 'ID of the receiveDataThread is %s. The size of the shared.connectedHostsList is now', + str(id(self)), len(shared.connectedHostsList)) while True: try: self.data += self.sock.recv(4096) except socket.timeout: - shared.printLock.acquire() - print 'Timeout occurred waiting for data from', self.HOST + '. Closing receiveData thread. (ID:', str(id(self)) + ')' - shared.printLock.release() + logger.info('Timeout occurred waiting for data from %s. Closing receiveData thread. (ID:%s)', + self.HOST, id(self)) break except Exception as err: - shared.printLock.acquire() - print 'sock.recv error. Closing receiveData thread (HOST:', self.HOST, 'ID:', str(id(self)) + ').', err - shared.printLock.release() + logger.exception('sock.recv error. Closing receiveData thread (HOST:%s, ID:%s).', + self.HOST, id(self)) break - # print 'Received', repr(self.data) + logger.debug('Received data: %s.' % repr(self.data)) if self.data == "": - shared.printLock.acquire() - print 'Connection to', self.HOST, 'closed. Closing receiveData thread. (ID:', str(id(self)) + ')' - shared.printLock.release() + logger.info('Connection to %s closed. Closing receiveData thread. (ID:%s)', self.HOST, id(self)) break else: self.processData() try: del selfInitiatedConnections[self.streamNumber][self] - shared.printLock.acquire() - print 'removed self (a receiveDataThread) from selfInitiatedConnections' - shared.printLock.release() + logger.info('Removed self (a receiveDataThread) from selfInitiatedConnections') except: pass shared.broadcastToSendDataQueues((0, 'shutdown', self.HOST)) try: del shared.connectedHostsList[self.HOST] except Exception as err: - shared.printLock.acquire() - print 'Could not delete', self.HOST, 'from shared.connectedHostsList.', err - shared.printLock.release() + logger.exception('Could not delete from shared.connectedHostsList.' % (self.host)) try: del numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer[ self.HOST] except: pass shared.UISignalQueue.put(('updateNetworkStatusTab', 'no data')) - shared.printLock.acquire() - print 'The size of the connectedHostsList is now:', len(shared.connectedHostsList) - shared.printLock.release() + logger.info('The size of the connectedHostsList is now: %s' % len(shared.connectedHostsList)) def processData(self): - global verbose - # if verbose >= 3: - # shared.printLock.acquire() - # print 'self.data is currently ', repr(self.data) - # shared.printLock.release() + logger.debug('self.data is currently %s', self.data) if len(self.data) < 20: # if so little of the data has arrived that we can't even unpack the payload length return if self.data[0:4] != '\xe9\xbe\xb4\xd9': - if verbose >= 1: - shared.printLock.acquire() - print 'The magic bytes were not correct. First 40 bytes of data: ' + repr(self.data[0:40]) - shared.printLock.release() + logger.debug('The magic bytes were not correct. First 40 bytes of data: ' + repr(self.data[0:40])) self.data = "" return self.payloadLength, = unpack('>L', self.data[16:20]) if len(self.data) < self.payloadLength + 24: # check if the whole message has arrived yet. 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... - print 'Checksum incorrect. Clearing this message.' + # 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]: + logger.info('Checksum incorrect. Clearing this message.') self.data = self.data[self.payloadLength + 24:] self.processData() return + # The time we've last seen this node is obviously right now since we # just received valid data from it. So update the knownNodes list so # that other peers can be made aware of its existance. - if self.initiatedConnection and self.connectionIsOrWasFullyEstablished: # The remote port is only something we should share with others if it is the remote node's incoming port (rather than some random operating-system-assigned outgoing port). + # + # The remote port is only something we should share with others if it is the remote node's incoming port (rather + # than some random operating-system-assigned outgoing port). + if self.initiatedConnection and self.connectionIsOrWasFullyEstablished: shared.knownNodesLock.acquire() shared.knownNodes[self.streamNumber][ self.HOST] = (self.PORT, int(time.time())) 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 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: remoteCommand = self.data[4:16] - shared.printLock.acquire() - print 'remoteCommand', repr(remoteCommand.replace('\x00', '')), ' from', self.HOST - shared.printLock.release() + logger.warning('remoteCommand %s from %s', remoteCommand.replace('\x00', ''), self.HOST) if remoteCommand == 'version\x00\x00\x00\x00\x00': self.recversion(self.data[24:self.payloadLength + 24]) elif remoteCommand == 'verack\x00\x00\x00\x00\x00\x00': From 521c800168f7588916d6bdc5c04635c8972c9983 Mon Sep 17 00:00:00 2001 From: Gregor Robinson Date: Thu, 20 Jun 2013 13:47:52 -0600 Subject: [PATCH 2/6] Continue migration from print messages This is continuing xj9's work to migrate from print-based logging to Python logger-based log messages. This still has a long way to go. This also also has some minor formatting changes for readability, especially for those of us not coding on an enormous HD monitor. (Forgive me!) --- src/bitmessagemain.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/bitmessagemain.py b/src/bitmessagemain.py index b4cf7da7..a8b6426c 100644 --- a/src/bitmessagemain.py +++ b/src/bitmessagemain.py @@ -8,7 +8,7 @@ # yet contain logic to expand into further streams. # The software version variable is now held in shared.py -verbose=2 +verbose=1 maximumAgeOfAnObjectThatIAmWillingToAccept = 216000 # Equals two days and 12 hours. lengthOfTimeToLeaveObjectsInInventory = 237600 # Equals two days and 18 hours. This should be longer than maximumAgeOfAnObjectThatIAmWillingToAccept so that we don't process messages twice. lengthOfTimeToHoldOnToAllPubkeys = 2419200 # Equals 4 weeks. You could make this longer if you want but making it shorter would not be advisable because there is a very small possibility that it could keep you from obtaining a needed pubkey for a period of time. From dcb29d0c5bc2bae74bbab0f5ff5cb629a0af9637 Mon Sep 17 00:00:00 2001 From: Gregor Robinson Date: Thu, 20 Jun 2013 13:53:41 -0600 Subject: [PATCH 3/6] Undoing stupid unintentional formatting change. --- src/bitmessagemain.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/bitmessagemain.py b/src/bitmessagemain.py index a8b6426c..7b6e3f17 100644 --- a/src/bitmessagemain.py +++ b/src/bitmessagemain.py @@ -8,7 +8,7 @@ # yet contain logic to expand into further streams. # The software version variable is now held in shared.py -verbose=1 +verbose = 1 maximumAgeOfAnObjectThatIAmWillingToAccept = 216000 # Equals two days and 12 hours. lengthOfTimeToLeaveObjectsInInventory = 237600 # Equals two days and 18 hours. This should be longer than maximumAgeOfAnObjectThatIAmWillingToAccept so that we don't process messages twice. lengthOfTimeToHoldOnToAllPubkeys = 2419200 # Equals 4 weeks. You could make this longer if you want but making it shorter would not be advisable because there is a very small possibility that it could keep you from obtaining a needed pubkey for a period of time. From 0f0ff3ae6b66d86cf8bcdc0be2fbb7c2f63d9e9e Mon Sep 17 00:00:00 2001 From: Gregor Robinson Date: Thu, 20 Jun 2013 14:38:05 -0600 Subject: [PATCH 4/6] Don't log self.data Too much data to print to screen except, perhaps, in exceptional circumstances. --- src/bitmessagemain.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/bitmessagemain.py b/src/bitmessagemain.py index 1d20a97d..9f32ab76 100644 --- a/src/bitmessagemain.py +++ b/src/bitmessagemain.py @@ -326,7 +326,6 @@ class receiveDataThread(threading.Thread): logger.exception('sock.recv error. Closing receiveData thread (HOST:%s, ID:%s).', self.HOST, id(self)) break - logger.debug('Received data: %s.' % repr(self.data)) if self.data == "": logger.info('Connection to %s closed. Closing receiveData thread. (ID:%s)', self.HOST, id(self)) break @@ -352,7 +351,6 @@ class receiveDataThread(threading.Thread): logger.info('The size of the connectedHostsList is now: %s' % len(shared.connectedHostsList)) def processData(self): - logger.debug('self.data is currently %s', self.data) if len(self.data) < 20: # if so little of the data has arrived that we can't even unpack the payload length return if self.data[0:4] != '\xe9\xbe\xb4\xd9': From 6d11e5a680c8b89dcdf82f96febdc77ad30a4926 Mon Sep 17 00:00:00 2001 From: Gregor Robinson Date: Thu, 20 Jun 2013 21:57:24 -0600 Subject: [PATCH 5/6] More print -> logger translations. --- src/bitmessagemain.py | 157 ++++++++++++++++++++---------------------- 1 file changed, 76 insertions(+), 81 deletions(-) diff --git a/src/bitmessagemain.py b/src/bitmessagemain.py index 9f32ab76..09ab7209 100644 --- a/src/bitmessagemain.py +++ b/src/bitmessagemain.py @@ -313,7 +313,7 @@ class receiveDataThread(threading.Thread): def run(self): logger.info( - 'ID of the receiveDataThread is %s. The size of the shared.connectedHostsList is now', + 'ID of the receiveDataThread is %s. The size of the shared.connectedHostsList is now %s', str(id(self)), len(shared.connectedHostsList)) while True: try: @@ -334,7 +334,7 @@ class receiveDataThread(threading.Thread): try: del selfInitiatedConnections[self.streamNumber][self] - logger.info('Removed self (a receiveDataThread) from selfInitiatedConnections') + logger.debug('Removed self (a receiveDataThread) from selfInitiatedConnections') except: pass shared.broadcastToSendDataQueues((0, 'shutdown', self.HOST)) @@ -385,7 +385,7 @@ class receiveDataThread(threading.Thread): # systems are less tolarant of large messages.) if self.payloadLength <= 180000000: remoteCommand = self.data[4:16] - logger.warning('remoteCommand %s from %s', remoteCommand.replace('\x00', ''), self.HOST) + logger.info('remoteCommand %s from %s', remoteCommand.replace('\x00', ''), self.HOST) if remoteCommand == 'version\x00\x00\x00\x00\x00': self.recversion(self.data[24:self.payloadLength + 24]) elif remoteCommand == 'verack\x00\x00\x00\x00\x00\x00': @@ -419,26 +419,24 @@ class receiveDataThread(threading.Thread): objectHash, = random.sample( self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave, 1) if objectHash in shared.inventory: - shared.printLock.acquire() - print 'Inventory (in memory) already has object listed in inv message.' - shared.printLock.release() + logger.debug('Inventory (in memory) already has object listed in inv message.') del self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave[ objectHash] elif isInSqlInventory(objectHash): if verbose >= 3: - shared.printLock.acquire() - print 'Inventory (SQL on disk) already has object listed in inv message.' - shared.printLock.release() + logger.debug('Inventory (SQL on disk) already has object listed in inv message.') del self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave[ objectHash] else: self.sendgetdata(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. 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] if len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) == 0: - shared.printLock.acquire() - print '(concerning', self.HOST + ')', 'number of objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave is now', len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) - shared.printLock.release() + logger.debug('(concerning %s) number of ' + 'objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave is now %s', + self.HOST, len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave)) try: del numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer[ self.HOST] # 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. @@ -446,18 +444,18 @@ class receiveDataThread(threading.Thread): pass break if len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) == 0: - shared.printLock.acquire() - print '(concerning', self.HOST + ')', 'number of objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave is now', len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) - shared.printLock.release() + logger.debug('(concerning %s) number of objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave is now %s', + self.HOST, + len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave)) try: del numberOfObjectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHavePerPeer[ self.HOST] # 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. except: pass if len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) > 0: - shared.printLock.acquire() - print '(concerning', self.HOST + ')', 'number of objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave is now', len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) - shared.printLock.release() + logger.debug('(concerning %s) number of objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave is now %s', + self.HOST, + self.len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave)) 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. if len(self.ackDataThatWeHaveYetToSend) > 0: @@ -475,22 +473,18 @@ class receiveDataThread(threading.Thread): payloadLengthExtraBytes = shared.networkDefaultPayloadLengthExtraBytes POW, = unpack('>Q', hashlib.sha512(hashlib.sha512(data[ :8] + hashlib.sha512(data[8:]).digest()).digest()).digest()[0:8]) - # print 'POW:', POW return POW <= 2 ** 64 / ((len(data) + payloadLengthExtraBytes) * (nonceTrialsPerByte)) def sendpong(self): - print 'Sending pong' + logger.debug('Sending pong.') try: 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') except Exception as err: - # if not 'Bad file descriptor' in err: - shared.printLock.acquire() - sys.stderr.write('sock.sendall error: %s\n' % err) - shared.printLock.release() + logger.exception('Caught exception sending pong.') def recverack(self): - print 'verack received' + logger.debug('verack received') self.verackReceived = True if self.verackSent: # We have thus both sent and received a verack. @@ -505,19 +499,15 @@ class receiveDataThread(threading.Thread): shared.UISignalQueue.put(('updateNetworkStatusTab', 'no data')) remoteNodeIncomingPort, remoteNodeSeenTime = shared.knownNodes[ self.streamNumber][self.HOST] - shared.printLock.acquire() - print 'Connection fully established with', self.HOST, remoteNodeIncomingPort - print 'The size of the connectedHostsList is now', len(shared.connectedHostsList) - print 'The length of sendDataQueues is now:', len(shared.sendDataQueues) - print 'broadcasting addr from within connectionFullyEstablished function.' - shared.printLock.release() + logger.info('Connection fully established with %s:%s', self.HOST, remoteNodeIncomingPort) + logger.info('The size of the connectedHostsList is now: %s', len(shared.connectedHostsList)) + logger.info('The length of sendDataQueues is now: %s', len(shared.sendDataQueues)) + logger.info('broadcasting addr from within connectionFullyEstablished function.') self.broadcastaddr([(int(time.time()), self.streamNumber, 1, self.HOST, remoteNodeIncomingPort)]) # This lets all of our peers know about this new node. self.sendaddr() # This is one large addr message to this one peer. if not self.initiatedConnection and len(shared.connectedHostsList) > 200: - shared.printLock.acquire() - print 'We are connected to too many people. Closing connection.' - shared.printLock.release() + logger.warning('We are connected to too many people. Closing connection.') shared.broadcastToSendDataQueues((0, 'shutdown', self.HOST)) return self.sendBigInv() @@ -569,23 +559,18 @@ class receiveDataThread(threading.Thread): headerData += 'inv\x00\x00\x00\x00\x00\x00\x00\x00\x00' headerData += pack('>L', len(payload)) headerData += hashlib.sha512(payload).digest()[:4] - shared.printLock.acquire() - print 'Sending huge inv message with', numberOfObjects, 'objects to just this one peer' - shared.printLock.release() + logger.info('Sending huge inv message with %s objects to just this one peer.', numberOfObjects) try: self.sock.sendall(headerData + payload) except Exception as err: - # if not 'Bad file descriptor' in err: - shared.printLock.acquire() - sys.stderr.write('sock.sendall error: %s\n' % err) - shared.printLock.release() + logg.exception('Error sending inv.') # We have received a broadcast message def recbroadcast(self, data): self.messageProcessingStartTime = time.time() # First we must check to make sure the proof of work is sufficient. if not self.isProofOfWorkSufficient(data): - print 'Proof of work in broadcast message insufficient.' + logg.info('Proof of work in broadcast message insufficient.') return readPosition = 8 # bypass the nonce embeddedTime, = unpack('>I', data[readPosition:readPosition + 4]) @@ -599,13 +584,13 @@ class receiveDataThread(threading.Thread): readPosition += 4 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. That doesn\'t make sense. Ignoring message.') return if embeddedTime < (int(time.time()) - 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 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 # Let us check to make sure the stream number is correct (thus # preventing an individual from sending broadcasts out on the wrong @@ -616,17 +601,20 @@ class receiveDataThread(threading.Thread): streamNumber, streamNumberLength = decodeVarint(data[ readPosition + broadcastVersionLength:readPosition + broadcastVersionLength + 10]) 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.info('The stream number encoded in this broadcast message (%s) does not ' + 'match the stream number on which it was received. Ignoring it.', + streamNumber) return shared.inventoryLock.acquire() self.inventoryHash = calculateInventoryHash(data) if self.inventoryHash in shared.inventory: - print 'We have already received this broadcast object. Ignoring.' + logger.info('We have already received this broadcast object. Ignoring.') shared.inventoryLock.release() return elif isInSqlInventory(self.inventoryHash): - print 'We have already received this broadcast object (it is stored on disk in the SQL inventory). Ignoring it.' + logger.info('We have already received this broadcast object ' + '(it is stored on disk in the SQL inventory). Ignoring it.') shared.inventoryLock.release() return # It is valid so far. Let's let our peers know about it. @@ -668,7 +656,9 @@ class receiveDataThread(threading.Thread): data[readPosition:readPosition + 9]) readPosition += broadcastVersionLength if broadcastVersion < 1 or broadcastVersion > 2: - print 'Cannot decode incoming broadcast versions higher than 2. Assuming the sender isn\'t being silly, you should upgrade Bitmessage because this message shall be ignored.' + logger.error('Cannot decode incoming broadcast versions higher than 2. ' + 'Assuming the sender isn\'t being silly, you should upgrade ' + 'Bitmessage because this message shall be ignored.') return if broadcastVersion == 1: beginningOfPubkeyPosition = readPosition # used when we add the pubkey to our pubkey table @@ -760,9 +750,7 @@ class receiveDataThread(threading.Thread): fromAddress = encodeAddress( sendersAddressVersion, sendersStream, ripe.digest()) - shared.printLock.acquire() - print 'fromAddress:', fromAddress - shared.printLock.release() + logger.debug('fromAddress: %s', fromAddress) if messageEncodingType == 2: bodyPositionIndex = string.find(message, '\nBody:') if bodyPositionIndex > 1: @@ -775,7 +763,7 @@ class receiveDataThread(threading.Thread): body = message subject = '' elif messageEncodingType == 0: - print 'messageEncodingType == 0. Doing nothing with the message.' + logger.debug('messageEncodingType == 0. Doing nothing with the message.') else: body = 'Unknown encoding type.\n\n' + repr(message) subject = '' @@ -807,29 +795,32 @@ class receiveDataThread(threading.Thread): call([apiNotifyPath, "newBroadcast"]) # Display timing data - shared.printLock.acquire() - print 'Time spent processing this interesting broadcast:', time.time() - self.messageProcessingStartTime - shared.printLock.release() + logger.info('Time spent processing this interesting broadcast: %s', + time.time() - self.messageProcessingStartTime) if broadcastVersion == 2: cleartextStreamNumber, cleartextStreamNumberLength = decodeVarint( data[readPosition:readPosition + 10]) readPosition += cleartextStreamNumberLength initialDecryptionSuccessful = False for key, cryptorObject in shared.MyECSubscriptionCryptorObjects.items(): + # TODO(grobinson): Once successful, shouldn't keep trying. try: decryptedData = cryptorObject.decrypt(data[readPosition:]) - toRipe = key # This is the RIPE hash of the sender's pubkey. We need this below to compare to the RIPE hash of the sender's address to verify that it was encrypted by with their key rather than some other key. + # This is the RIPE hash of the sender's pubkey. We need this below to compare to + # the RIPE hash of the sender's address to verify that it was encrypted by with + # their key rather than some other key. + toRipe = key initialDecryptionSuccessful = True - print 'EC decryption successful using key associated with ripe hash:', key.encode('hex') + logger.info('EC decryption successful using key associated with ripe hash:', + key.encode('hex')) break except Exception as err: + # TODO(fiatflux): avoid relying on exceptions for normal operation. pass - # print 'cryptorObject.decrypt Exception:', err if not initialDecryptionSuccessful: # This is not a broadcast I am interested in. - shared.printLock.acquire() - print 'Length of time program spent failing to decrypt this v2 broadcast:', time.time() - self.messageProcessingStartTime, 'seconds.' - shared.printLock.release() + logger.info('Time spent failing to decrypt this v2 broadcast: %s seconds.', + time.time() - self.messageProcessingStartTime) return # At this point this is a broadcast I have decrypted and thus am # interested in. @@ -839,13 +830,17 @@ class receiveDataThread(threading.Thread): sendersAddressVersion, sendersAddressVersionLength = decodeVarint( decryptedData[readPosition:readPosition + 9]) if sendersAddressVersion < 2 or sendersAddressVersion > 3: - print 'Cannot decode senderAddressVersion other than 2 or 3. Assuming the sender isn\'t being silly, you should upgrade Bitmessage because this message shall be ignored.' + logger.error('Cannot decode senderAddressVersion other than 2 or 3. Assuming the ' + 'sender isn\'t being silly, you should upgrade Bitmessage because ' + 'this message shall be ignored.') return readPosition += sendersAddressVersionLength sendersStream, sendersStreamLength = decodeVarint( decryptedData[readPosition:readPosition + 9]) if sendersStream != cleartextStreamNumber: - print 'The stream number outside of the encryption on which the POW was completed doesn\'t match the stream number inside the encryption. Ignoring broadcast.' + logger.info('The stream number outside of the encryption on which the POW was ' + 'completed doesn\'t match the stream number inside the encryption. ' + 'Ignoring broadcast.') return readPosition += sendersStreamLength behaviorBitfield = decryptedData[readPosition:readPosition + 4] @@ -860,11 +855,13 @@ class receiveDataThread(threading.Thread): requiredAverageProofOfWorkNonceTrialsPerByte, varintLength = decodeVarint( decryptedData[readPosition:readPosition + 10]) readPosition += varintLength - print 'sender\'s requiredAverageProofOfWorkNonceTrialsPerByte is', requiredAverageProofOfWorkNonceTrialsPerByte + logger.debug('Sender\'s requiredAverageProofOfWorkNonceTrialsPerByte is %s', + requiredAverageProofOfWorkNonceTrialsPerByte) requiredPayloadLengthExtraBytes, varintLength = decodeVarint( decryptedData[readPosition:readPosition + 10]) readPosition += varintLength - print 'sender\'s requiredPayloadLengthExtraBytes is', requiredPayloadLengthExtraBytes + logger.debug('sender\'s requiredPayloadLengthExtraBytes is %s', + requiredPayloadLengthExtraBytes) endOfPubkeyPosition = readPosition sha = hashlib.new('sha512') @@ -873,7 +870,8 @@ class receiveDataThread(threading.Thread): ripe.update(sha.digest()) if toRipe != ripe.digest(): - print 'The encryption key used to encrypt this message doesn\'t match the keys inbedded in the message itself. Ignoring message.' + logger.warning('The encryption key used to encrypt this message doesn\'t match ' + 'keys inbedded in the message itself. Ignoring message.') return messageEncodingType, messageEncodingTypeLength = decodeVarint( decryptedData[readPosition:readPosition + 9]) @@ -893,11 +891,11 @@ class receiveDataThread(threading.Thread): readPosition:readPosition + signatureLength] try: if not highlevelcrypto.verify(decryptedData[:readPositionAtBottomOfMessage], signature, sendersPubSigningKey.encode('hex')): - print 'ECDSA verify failed' + logger.warning('ECDSA verify failed.') return - print 'ECDSA verify passed' + logger.debug('ECDSA verify passed.') except Exception as err: - print 'ECDSA verify failed', err + logger.exception('ECDSA verify failed') return # verify passed @@ -920,9 +918,7 @@ class receiveDataThread(threading.Thread): fromAddress = encodeAddress( sendersAddressVersion, sendersStream, ripe.digest()) - shared.printLock.acquire() - print 'fromAddress:', fromAddress - shared.printLock.release() + logger.info('fromAddress: %s.', fromAddress) if messageEncodingType == 2: bodyPositionIndex = string.find(message, '\nBody:') if bodyPositionIndex > 1: @@ -935,7 +931,7 @@ class receiveDataThread(threading.Thread): body = message subject = '' elif messageEncodingType == 0: - print 'messageEncodingType == 0. Doing nothing with the message.' + logger.info('messageEncodingType == 0. Doing nothing with the message.') else: body = 'Unknown encoding type.\n\n' + repr(message) subject = '' @@ -967,16 +963,15 @@ class receiveDataThread(threading.Thread): call([apiNotifyPath, "newBroadcast"]) # Display timing data - shared.printLock.acquire() - print 'Time spent processing this interesting broadcast:', time.time() - self.messageProcessingStartTime - shared.printLock.release() + logger.info('Time spent processing this interesting broadcast:', + time.time() - self.messageProcessingStartTime) # We have received a msg message. def recmsg(self, data): self.messageProcessingStartTime = time.time() # First we must check to make sure the proof of work is sufficient. if not self.isProofOfWorkSufficient(data): - print 'Proof of work in msg message insufficient.' + logger.info('Proof of work in msg message insufficient.') return readPosition = 8 @@ -991,10 +986,10 @@ class receiveDataThread(threading.Thread): readPosition += 4 if embeddedTime > int(time.time()) + 10800: - print 'The time in the msg message is too new. Ignoring it. Time:', embeddedTime + logger.warning('The time in the msg message is too new. Ignoring it. Time: %s', embeddedTime) return if embeddedTime < int(time.time()) - maximumAgeOfAnObjectThatIAmWillingToAccept: - print 'The time in the msg message is too old. Ignoring it. Time:', embeddedTime + logger.warning('The time in the msg message is too old. Ignoring it. Time:', embeddedTime) return streamNumberAsClaimedByMsg, streamNumberAsClaimedByMsgLength = decodeVarint( data[readPosition:readPosition + 9]) From c1a30d3f54790214f63657601c62f02f44d22e67 Mon Sep 17 00:00:00 2001 From: Gregor Robinson Date: Thu, 20 Jun 2013 22:54:08 -0600 Subject: [PATCH 6/6] Fix typos introduced by 6d11e5a. --- src/bitmessagemain.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/bitmessagemain.py b/src/bitmessagemain.py index 6daa385f..62809837 100755 --- a/src/bitmessagemain.py +++ b/src/bitmessagemain.py @@ -457,7 +457,7 @@ class receiveDataThread(threading.Thread): if len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave) > 0: logger.debug('(concerning %s) number of objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave is now %s', self.HOST, - self.len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave)) + len(self.objectsThatWeHaveYetToCheckAndSeeWhetherWeAlreadyHave)) 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. if len(self.ackDataThatWeHaveYetToSend) > 0: @@ -752,7 +752,7 @@ class receiveDataThread(threading.Thread): fromAddress = encodeAddress( sendersAddressVersion, sendersStream, ripe.digest()) - logger.info('fromAddress: %s' str(fromAddress)) + logger.info('fromAddress: %s', str(fromAddress)) if messageEncodingType == 2: bodyPositionIndex = string.find(message, '\nBody:') if bodyPositionIndex > 1: @@ -813,11 +813,13 @@ class receiveDataThread(threading.Thread): # their key rather than some other key. toRipe = key initialDecryptionSuccessful = True - logger.info('EC decryption successful using key associated with ripe hash:', + logger.info('EC decryption successful using key associated with ripe hash: %s', key.encode('hex')) break except Exception as err: - logger.exception('cryptorObject.decrypt Exception: %s' % str(err)) + # TODO(grobinson): Avoid using exceptions for normal + # application logic. + pass if not initialDecryptionSuccessful: # This is not a broadcast I am interested in. logger.info('Time spent failing to decrypt this v2 broadcast: %s seconds.',