Asyncore updates

- mainly work on proxy support, but it's still not fully working - minor bugfixes
Bitmessage · Jun 10, 2017 · cba7490 · cba7490
1 parent 7deb7c3
commit cba7490
Show file tree

Hide file tree

Showing 9 changed files with 53 additions and 21 deletions.
diff --git a/src/network/asyncore_pollchoose.py b/src/network/asyncore_pollchoose.py
@@ -57,7 +57,7 @@
 import os
 from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, EINVAL, \
      ENOTCONN, ESHUTDOWN, EISCONN, EBADF, ECONNABORTED, EPIPE, EAGAIN, \
-     ECONNREFUSED, EHOSTUNREACH, ENETUNREACH, ENOTSOCK, \
+     ECONNREFUSED, EHOSTUNREACH, ENETUNREACH, ENOTSOCK, EINTR, \
      errorcode
 try:
     from errno import WSAEWOULDBLOCK

diff --git a/src/network/bmproto.py b/src/network/bmproto.py
@@ -282,11 +282,11 @@ def bm_command_object(self):
         except (BMObjectExpiredError, BMObjectUnwantedStreamError):
             for connection in network.connectionpool.BMConnectionPool().inboundConnections.values() + network.connectionpool.BMConnectionPool().outboundConnections.values():
                 try:
-                    del connection.objectsNewtoThem[hashId]
+                    del connection.objectsNewtoThem[self.object.inventoryHash]
                 except KeyError:
                     pass
                 try:
-                    del connection.objectsNewToMe[hashId]
+                    del connection.objectsNewToMe[self.object.inventoryHash]
                 except KeyError:
                     pass
             if not BMConfigParser().get("inventory", "acceptmismatch"):
@@ -459,7 +459,10 @@ def assembleAddr(peerList):
     def handle_close(self, reason=None):
         self.set_state("close")
         if reason is None:
-            logger.debug("%s:%i: closing", self.destination.host, self.destination.port)
+            try:
+                logger.debug("%s:%i: closing", self.destination.host, self.destination.port)
+            except AttributeError:
+                logger.debug("Disconnected socket closing")
         else:
             logger.debug("%s:%i: closing, %s", self.destination.host, self.destination.port, reason)
         AdvancedDispatcher.handle_close(self)
diff --git a/src/network/connectionpool.py b/src/network/connectionpool.py
@@ -7,6 +7,7 @@
 from bmconfigparser import BMConfigParser
 from debug import logger
 import helper_bootstrap
+from network.proxy import Proxy
 import network.bmproto
 import network.tcp
 import network.udp
@@ -129,6 +130,8 @@ def loop(self):
             if not self.bootstrapped:
                 helper_bootstrap.dns()
                 self.bootstrapped = True
+                Proxy.proxy = (BMConfigParser().safeGet("bitmessagesettings", "sockshostname"),
+                        BMConfigParser().safeGetInt("bitmessagesettings", "socksport"))
             established = sum(1 for c in self.outboundConnections.values() if (c.connected and c.fullyEstablished))
             pending = len(self.outboundConnections) - established
             if established < BMConfigParser().safeGetInt("bitmessagesettings", "maxoutboundconnections"):

diff --git a/src/network/objectracker.py b/src/network/objectracker.py
@@ -54,6 +54,7 @@ def initAddrBloom(self):
     def clean(self):
         if self.lastCleaned < time.time() - ObjectTracker.invCleanPeriod:
             if haveBloom:
+                # FIXME
                 if PendingDownloadQueue().size() == 0:
                     self.initInvBloom()
                 self.initAddrBloom()

diff --git a/src/network/proxy.py b/src/network/proxy.py
@@ -1,7 +1,10 @@
 import socket
 
+import state
+
 from advanceddispatcher import AdvancedDispatcher
 import asyncore_pollchoose as asyncore
+import network.connectionpool
 
 class ProxyError(Exception): pass
 class GeneralProxyError(ProxyError): pass
@@ -32,10 +35,25 @@ def auth(self, authTuple):
         self.__class__._auth = authTuple
 
     def __init__(self, address):
-        if type(address) != tuple or (len(address) < 2) or (type(str(address[0])) != type('')) or (type(address[1]) != int):
+        if not isinstance(address, state.Peer):
             raise ValueError
         AdvancedDispatcher.__init__(self)
         self.destination = address
+        self.isOutbound = True
         self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
         self.connect(self.proxy)
         print "connecting in background to %s:%i" % (self.proxy[0], self.proxy[1])
+
+    def handle_connect(self):
+        try:
+            AdvancedDispatcher.handle_connect(self)
+        except socket.error as e:
+            if e.errno in asyncore._DISCONNECTED:
+                logger.debug("%s:%i: Connection failed: %s" % (self.destination.host, self.destination.port, str(e)))
+                return
+
+    def state_proxy_handshake_done(self):
+        self.writeQueue.put(protocol.assembleVersionMessage(self.destination.host, self.destination.port, network.connectionpool.BMConnectionPool().streams, False))
+        self.connectedAt = time.time()
+        return False
+
diff --git a/src/network/socks4a.py b/src/network/socks4a.py
@@ -43,7 +43,7 @@ def state_pre_connect(self):
             self.__proxypeername = (socket.inet_ntoa(self.ipaddr), self.destination[1])
         else:
             self.__proxypeername = (self.destination[0], self.destport)
-        self.set_state("socks_handshake_done", 8)
+        self.set_state("proxy_handshake_done", 8)
 
     def proxy_sock_name(self):
        return socket.inet_ntoa(self.__proxysockname[0])

diff --git a/src/network/socks5.py b/src/network/socks5.py
@@ -4,6 +4,7 @@
 from advanceddispatcher import AdvancedDispatcher
 import asyncore_pollchoose as asyncore
 from proxy import Proxy, ProxyError, GeneralProxyError
+import network.connectionpool
 
 class Socks5AuthError(ProxyError): pass
 class Socks5Error(ProxyError): pass
@@ -103,7 +104,7 @@ def state_proxy_addr_2_1(self):
     def state_proxy_addr_2_2(self):
         if not self.read_buf_sufficient(self.address_length):
             return False
-        self.boundaddr = read_buf
+        self.boundaddr = self.read_buf
         self.set_state("proxy_port", self.address_length)
 
     def state_proxy_port(self):
@@ -115,14 +116,11 @@ def state_proxy_port(self):
             self.__proxypeername = (socket.inet_ntoa(self.ipaddr), self.destination[1])
         else:
             self.__proxypeername = (self.destination[0], self.destport)
-        self.set_state("socks_handshake_done", 2)
+        self.set_state("proxy_handshake_done", 2)
 
     def proxy_sock_name(self):
        return socket.inet_ntoa(self.__proxysockname[0])
 
-    def state_socks_handshake_done(self):
-        return False
-
 
 class Socks5Connection(Socks5):
     def __init__(self, address):

diff --git a/src/network/tcp.py b/src/network/tcp.py
@@ -49,6 +49,10 @@ def __init__(self, address=None, sock=None):
             TLSDispatcher.__init__(self, sock, server_side=True)
             self.connectedAt = time.time()
             logger.debug("Received connection from %s:%i", self.destination.host, self.destination.port)
+        elif address is not None and sock is not None:
+            TLSDispatcher.__init__(self, sock, server_side=False)
+            self.isOutbound = True
+            logger.debug("Outbound proxy connection to %s:%i", self.destination.host, self.destination.port)
         else:
             self.destination = address
             self.isOutbound = True
@@ -159,33 +163,37 @@ def handle_connect(self):
         self.connectedAt = time.time()
 
     def handle_read(self):
-        try:
-            TLSDispatcher.handle_read(self)
-        except socket.error as e:
-            logger.debug("%s:%i: Handle read fail: %s" % (self.destination.host, self.destination.port, str(e)))
+#        try:
+        TLSDispatcher.handle_read(self)
+#        except socket.error as e:
+#            logger.debug("%s:%i: Handle read fail: %s" % (self.destination.host, self.destination.port, str(e)))
 
     def handle_write(self):
-        try:
-            TLSDispatcher.handle_write(self)
-        except socket.error as e:
-            logger.debug("%s:%i: Handle write fail: %s" % (self.destination.host, self.destination.port, str(e)))
+#        try:
+        TLSDispatcher.handle_write(self)
+#        except socket.error as e:
+#            logger.debug("%s:%i: Handle write fail: %s" % (self.destination.host, self.destination.port, str(e)))
 
 
 class Socks5BMConnection(Socks5Connection, TCPConnection):
     def __init__(self, address):
         Socks5Connection.__init__(self, address=address)
+        TCPConnection.__init__(self, address=address, sock=self.socket)
+        self.set_state("init")
 
     def state_socks_handshake_done(self):
-        TCPConnection.state_init(self)
+        self.set_state("bm_header", expectBytes=protocol.Header.size)
         return False
 
 
 class Socks4aBMConnection(Socks4aConnection, TCPConnection):
     def __init__(self, address):
         Socks4aConnection.__init__(self, address=address)
+        TCPConnection.__init__(self, address=address, sock=self.socket)
+        self.set_state("init")
 
     def state_socks_handshake_done(self):
-        TCPConnection.state_init(self)
+        self.set_state("bm_header", expectBytes=protocol.Header.size)
         return False
 
 

diff --git a/src/network/udp.py b/src/network/udp.py
@@ -3,6 +3,7 @@
 import hashlib
 import math
 import time
+import Queue
 import socket
 import struct
 import random