2013-06-22 01:49:50 +02:00
import threading
import time
import random
import shared
2015-11-22 16:18:59 +01:00
import select
2013-06-22 01:49:50 +02:00
import socks
import socket
import sys
2013-06-24 21:51:01 +02:00
import tr
2013-06-22 01:49:50 +02:00
from class_sendDataThread import *
from class_receiveDataThread import *
2015-11-22 16:18:59 +01:00
from helper_threading import *
2013-06-22 01:49:50 +02:00
# For each stream to which we connect, several outgoingSynSender threads
# will exist and will collectively create 8 connections with peers.
2015-11-22 16:18:59 +01:00
class outgoingSynSender ( threading . Thread , StoppableThread ) :
2013-06-22 01:49:50 +02:00
def __init__ ( self ) :
2015-11-18 16:22:17 +01:00
threading . Thread . __init__ ( self , name = " outgoingSynSender " )
2015-11-22 16:18:59 +01:00
self . initStop ( )
2013-06-22 01:49:50 +02:00
def setup ( self , streamNumber , selfInitiatedConnections ) :
self . streamNumber = streamNumber
self . selfInitiatedConnections = selfInitiatedConnections
2014-02-06 14:16:07 +01:00
def _getPeer ( self ) :
# If the user has specified a trusted peer then we'll only
# ever connect to that. Otherwise we'll pick a random one from
# the known nodes
shared . knownNodesLock . acquire ( )
if shared . trustedPeer :
peer = shared . trustedPeer
shared . knownNodes [ self . streamNumber ] [ peer ] = time . time ( )
else :
peer , = random . sample ( shared . knownNodes [ self . streamNumber ] , 1 )
shared . knownNodesLock . release ( )
return peer
2015-11-22 16:18:59 +01:00
def stopThread ( self ) :
super ( outgoingSynSender , self ) . stopThread ( )
try :
self . sock . shutdown ( socket . SHUT_RDWR )
except :
pass
2014-02-06 14:16:07 +01:00
2013-06-22 01:49:50 +02:00
def run ( self ) :
2015-11-22 16:18:59 +01:00
while shared . safeConfigGetBoolean ( ' bitmessagesettings ' , ' dontconnect ' ) and not self . _stopped :
self . stop . wait ( 2 )
while shared . safeConfigGetBoolean ( ' bitmessagesettings ' , ' sendoutgoingconnections ' ) and not self . _stopped :
2015-11-18 16:22:17 +01:00
self . name = " outgoingSynSender "
2014-02-06 14:16:07 +01:00
maximumConnections = 1 if shared . trustedPeer else 8 # maximum number of outgoing connections = 8
while len ( self . selfInitiatedConnections [ self . streamNumber ] ) > = maximumConnections :
2015-11-22 16:18:59 +01:00
self . stop . wait ( 10 )
2013-06-22 01:49:50 +02:00
if shared . shutdown :
break
random . seed ( )
2014-02-06 14:16:07 +01:00
peer = self . _getPeer ( )
2013-06-24 21:51:01 +02:00
shared . alreadyAttemptedConnectionsListLock . acquire ( )
2013-08-01 18:16:31 +02:00
while peer in shared . alreadyAttemptedConnectionsList or peer . host in shared . connectedHostsList :
2013-06-24 21:51:01 +02:00
shared . alreadyAttemptedConnectionsListLock . release ( )
2013-06-22 01:49:50 +02:00
# print 'choosing new sample'
random . seed ( )
2014-02-06 14:16:07 +01:00
peer = self . _getPeer ( )
2015-11-22 16:18:59 +01:00
self . stop . wait ( 1 )
if shared . shutdown :
break
2013-06-24 21:51:01 +02:00
# Clear out the shared.alreadyAttemptedConnectionsList every half
2013-06-22 01:49:50 +02:00
# hour so that this program will again attempt a connection
# to any nodes, even ones it has already tried.
2013-06-24 21:51:01 +02:00
if ( time . time ( ) - shared . alreadyAttemptedConnectionsListResetTime ) > 1800 :
shared . alreadyAttemptedConnectionsList . clear ( )
shared . alreadyAttemptedConnectionsListResetTime = int (
2013-06-22 01:49:50 +02:00
time . time ( ) )
2013-06-24 21:51:01 +02:00
shared . alreadyAttemptedConnectionsListLock . acquire ( )
2013-07-30 22:23:18 +02:00
shared . alreadyAttemptedConnectionsList [ peer ] = 0
2015-11-23 00:03:49 +01:00
try :
shared . alreadyAttemptedConnectionsListLock . release ( )
2015-11-23 00:05:49 +01:00
except threading . ThreadError as e :
2015-11-23 00:03:49 +01:00
pass
2015-11-18 16:22:17 +01:00
self . name = " outgoingSynSender- " + peer . host
2014-02-16 17:21:20 +01:00
if peer . host . find ( ' : ' ) == - 1 :
address_family = socket . AF_INET
else :
address_family = socket . AF_INET6
2014-04-28 00:05:43 +02:00
try :
2015-11-22 16:18:59 +01:00
self . sock = socks . socksocket ( address_family , socket . SOCK_STREAM )
2014-04-28 00:05:43 +02:00
except :
"""
The line can fail on Windows systems which aren ' t
64 - bit compatiable :
File " C: \ Python27 \ lib \ socket.py " , line 187 , in __init__
_sock = _realsocket ( family , type , proto )
error : [ Errno 10047 ] An address incompatible with the requested protocol was used
So let us remove the offending address from our knownNodes file .
"""
shared . knownNodesLock . acquire ( )
2014-09-10 22:47:51 +02:00
try :
del shared . knownNodes [ self . streamNumber ] [ peer ]
except :
pass
2014-04-28 00:05:43 +02:00
shared . knownNodesLock . release ( )
2015-11-18 16:22:17 +01:00
logger . debug ( ' deleting ' + str ( peer ) + ' from shared.knownNodes because it caused a socks.socksocket exception. We must not be 64-bit compatible. ' )
2014-04-28 00:05:43 +02:00
continue
2013-06-22 01:49:50 +02:00
# This option apparently avoids the TIME_WAIT state so that we
# can rebind faster
2015-11-22 16:18:59 +01:00
self . sock . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , 1 )
self . sock . settimeout ( 20 )
2013-06-24 21:51:01 +02:00
if shared . config . get ( ' bitmessagesettings ' , ' socksproxytype ' ) == ' none ' and shared . verbose > = 2 :
2015-11-18 16:22:17 +01:00
logger . debug ( ' Trying an outgoing connection to ' + str ( peer ) )
2013-06-29 19:29:35 +02:00
2013-06-22 01:49:50 +02:00
# sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
elif shared . config . get ( ' bitmessagesettings ' , ' socksproxytype ' ) == ' SOCKS4a ' :
2013-06-24 21:51:01 +02:00
if shared . verbose > = 2 :
2015-11-18 16:22:17 +01:00
logger . debug ( ' (Using SOCKS4a) Trying an outgoing connection to ' + str ( peer ) )
2013-06-29 19:29:35 +02:00
2013-06-22 01:49:50 +02:00
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.
if shared . config . getboolean ( ' bitmessagesettings ' , ' socksauthentication ' ) :
socksusername = shared . config . get (
' bitmessagesettings ' , ' socksusername ' )
sockspassword = shared . config . get (
' bitmessagesettings ' , ' sockspassword ' )
2015-11-22 16:18:59 +01:00
self . sock . setproxy (
2013-06-22 01:49:50 +02:00
proxytype , sockshostname , socksport , rdns , socksusername , sockspassword )
else :
2015-11-22 16:18:59 +01:00
self . sock . setproxy (
2013-06-22 01:49:50 +02:00
proxytype , sockshostname , socksport , rdns )
elif shared . config . get ( ' bitmessagesettings ' , ' socksproxytype ' ) == ' SOCKS5 ' :
2013-06-24 21:51:01 +02:00
if shared . verbose > = 2 :
2015-11-18 16:22:17 +01:00
logger . debug ( ' (Using SOCKS5) Trying an outgoing connection to ' + str ( peer ) )
2013-06-29 19:29:35 +02:00
2013-06-22 01:49:50 +02:00
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.
if shared . config . getboolean ( ' bitmessagesettings ' , ' socksauthentication ' ) :
socksusername = shared . config . get (
' bitmessagesettings ' , ' socksusername ' )
sockspassword = shared . config . get (
' bitmessagesettings ' , ' sockspassword ' )
2015-11-22 16:18:59 +01:00
self . sock . setproxy (
2013-06-22 01:49:50 +02:00
proxytype , sockshostname , socksport , rdns , socksusername , sockspassword )
else :
2015-11-22 16:18:59 +01:00
self . sock . setproxy (
2013-06-22 01:49:50 +02:00
proxytype , sockshostname , socksport , rdns )
try :
2015-11-22 16:18:59 +01:00
self . sock . connect ( ( peer . host , peer . port ) )
2013-06-22 01:49:50 +02:00
rd = receiveDataThread ( )
rd . daemon = True # close the main program even if there are threads left
2013-06-24 21:51:01 +02:00
someObjectsOfWhichThisRemoteNodeIsAlreadyAware = { } # This is not necessairly a complete list; we clear it from time to time to save memory.
2013-12-30 04:36:23 +01:00
sendDataThreadQueue = Queue . Queue ( ) # Used to submit information to the send data thread for this connection.
2015-11-22 16:18:59 +01:00
rd . setup ( self . sock ,
2013-12-30 04:36:23 +01:00
peer . host ,
peer . port ,
self . streamNumber ,
someObjectsOfWhichThisRemoteNodeIsAlreadyAware ,
self . selfInitiatedConnections ,
sendDataThreadQueue )
2013-06-22 01:49:50 +02:00
rd . start ( )
2015-11-18 16:22:17 +01:00
logger . debug ( str ( self ) + ' connected to ' + str ( peer ) + ' during an outgoing attempt. ' )
2013-06-29 19:29:35 +02:00
2013-06-22 01:49:50 +02:00
2013-12-30 04:36:23 +01:00
sd = sendDataThread ( sendDataThreadQueue )
2015-11-22 16:18:59 +01:00
sd . setup ( self . sock , peer . host , peer . port , self . streamNumber ,
2013-06-24 21:51:01 +02:00
someObjectsOfWhichThisRemoteNodeIsAlreadyAware )
2013-06-22 01:49:50 +02:00
sd . start ( )
sd . sendVersionMessage ( )
except socks . GeneralProxyError as err :
2013-06-24 21:51:01 +02:00
if shared . verbose > = 2 :
2015-11-18 16:22:17 +01:00
logger . debug ( ' Could NOT connect to ' + str ( peer ) + ' during outgoing attempt. ' + str ( err ) )
2013-06-29 19:29:35 +02:00
2014-09-10 22:47:51 +02:00
deletedPeer = None
with shared . knownNodesLock :
"""
It is remotely possible that peer is no longer in shared . knownNodes .
This could happen if two outgoingSynSender threads both try to
connect to the same peer , both fail , and then both try to remove
it from shared . knownNodes . This is unlikely because of the
alreadyAttemptedConnectionsList but because we clear that list once
every half hour , it can happen .
"""
if peer in shared . knownNodes [ self . streamNumber ] :
timeLastSeen = shared . knownNodes [ self . streamNumber ] [ peer ]
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.
del shared . knownNodes [ self . streamNumber ] [ peer ]
deletedPeer = peer
if deletedPeer :
2015-11-18 16:22:17 +01:00
str ( ' deleting ' + str ( peer ) + ' from shared.knownNodes because it is more than 48 hours old and we could not connect to it. ' )
2013-06-29 19:29:35 +02:00
2013-06-22 01:49:50 +02:00
except socks . Socks5AuthError as err :
shared . UISignalQueue . put ( (
2013-06-24 21:51:01 +02:00
' updateStatusBar ' , tr . translateText (
2013-06-22 01:49:50 +02:00
" MainWindow " , " SOCKS5 Authentication problem: % 1 " ) . arg ( str ( err ) ) ) )
except socks . Socks5Error as err :
pass
2015-11-18 16:22:17 +01:00
logger . error ( ' SOCKS5 error. (It is possible that the server wants authentication).) ' + str ( err ) )
2013-06-22 01:49:50 +02:00
except socks . Socks4Error as err :
2015-11-18 16:22:17 +01:00
logger . error ( ' Socks4Error: ' + str ( err ) )
2013-06-22 01:49:50 +02:00
except socket . error as err :
if shared . config . get ( ' bitmessagesettings ' , ' socksproxytype ' ) [ 0 : 5 ] == ' SOCKS ' :
2015-11-18 16:22:17 +01:00
logger . error ( ' Bitmessage MIGHT be having trouble connecting to the SOCKS server. ' + str ( err ) )
2013-06-22 01:49:50 +02:00
else :
2013-06-24 21:51:01 +02:00
if shared . verbose > = 1 :
2015-11-26 02:35:59 +01:00
logger . debug ( ' Could NOT connect to ' + str ( peer ) + ' during outgoing attempt. ' + str ( err ) )
2013-06-29 19:29:35 +02:00
2014-09-10 22:47:51 +02:00
deletedPeer = None
with shared . knownNodesLock :
"""
It is remotely possible that peer is no longer in shared . knownNodes .
This could happen if two outgoingSynSender threads both try to
connect to the same peer , both fail , and then both try to remove
it from shared . knownNodes . This is unlikely because of the
alreadyAttemptedConnectionsList but because we clear that list once
every half hour , it can happen .
"""
if peer in shared . knownNodes [ self . streamNumber ] :
timeLastSeen = shared . knownNodes [ self . streamNumber ] [ peer ]
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.
del shared . knownNodes [ self . streamNumber ] [ peer ]
deletedPeer = peer
if deletedPeer :
2015-11-18 16:22:17 +01:00
logger . debug ( ' deleting ' + str ( peer ) + ' from shared.knownNodes because it is more than 48 hours old and we could not connect to it. ' )
2013-06-29 19:29:35 +02:00
2013-06-22 01:49:50 +02:00
except Exception as err :
import traceback
2015-11-18 16:22:17 +01:00
logger . exception ( ' An exception has occurred in the outgoingSynSender thread that was not caught by other exception types: ' )
2015-11-22 16:18:59 +01:00
self . stop . wait ( 0.1 )