Ticket #2406: client-bind.patch

xpra/client/mixins/network_state.py

@@ -9 +9 @@
 from collections import deque
 
 from xpra.os_util import monotonic_time, POSIX
-from xpra.util import envint, envbool, csv, typedict
+from xpra.util import envint, envbool, envfloat, csv, typedict
 from xpra.exit_codes import EXIT_TIMEOUT
 from xpra.net.packet_encoding import ALL_ENCODERS
 from xpra.net.compression import ALL_COMPRESSORS
+from xpra.net.bytestreams import log_new_connection
+from xpra.net.socket_util import create_sockets, add_listen_socket, accept_connection
+from xpra.net.net_util import get_network_caps
+from xpra.net.protocol import Protocol
 from xpra.client.mixins.stub_client_mixin import StubClientMixin
-from xpra.scripts.config import parse_with_unit
+from xpra.scripts.config import parse_with_unit, InitException
 from xpra.log import Logger
 
 log = Logger("network")
@@ -27 +31 @@
 LOG_INFO_RESPONSE = os.environ.get("XPRA_LOG_INFO_RESPONSE", "")
 AUTO_BANDWIDTH_PCT = envint("XPRA_AUTO_BANDWIDTH_PCT", 80)
 assert 1<AUTO_BANDWIDTH_PCT<=100, "invalid value for XPRA_AUTO_BANDWIDTH_PCT: %i" % AUTO_BANDWIDTH_PCT
+BIND_SOCKETS = envbool("XPRA_CLIENT_BIND_SOCKETS", True)
+SOCKET_TIMEOUT = envfloat("XPRA_CLIENT_SOCKET_TIMEOUT", "0.1")
+MAX_CONCURRENT_CONNECTIONS = envint("XPRA_MAX_CONCURRENT_CONNECTIONS", 100)
 
 
 """
@@ -68 +75 @@
         self.ping_echo_timers = {}
         self.ping_echo_timeout_timer = None
 
+        #sockets:
+        self.sockets = {}
+        self.socket_info = {}
+        self.socket_options = {}
+        self.socket_cleanup = []
+        self._potential_protocols = []
 
+
     def init(self, opts):
         self.pings = opts.pings
         self.bandwidth_limit = parse_with_unit("bandwidth-limit", opts.bandwidth_limit)
         self.bandwidth_detection = opts.bandwidth_detection
         bandwidthlog("init bandwidth_limit=%s", self.bandwidth_limit)
+        if BIND_SOCKETS:
+            def err(msg):
+                raise InitException(msg)
+            #don't create regular local sockets or udp sockets for now:
+            opts.bind = ()
+            opts.bind_udp = ()
+            self.sockets = create_sockets(opts, err)
+            log.error("sockets=%s", self.sockets)
 
+    def run(self):
+        self.start_listen_sockets()
 
     def cleanup(self):
         self.cancel_ping_timer()
         self.cancel_ping_echo_timers()
         self.cancel_ping_echo_timeout_timer()
+        self.cleanup_sockets()
 
+    def cleanup_sockets(self):
+        for c in self.socket_cleanup:
+            try:
+                c()
+            except Exception:
+                log.error("Error during socket cleanup", exc_info=True)
 
+    def start_listen_sockets(self):
+        for sock_def, options in self.sockets.items():
+            socktype, sock, info, _ = sock_def
+            log("start_listen_sockets() will add %s socket %s (%s)", socktype, sock, info)
+            self.socket_info[sock] = info
+            self.socket_options[sock] = options
+            self.idle_add(self.add_listen_socket, socktype, sock)
+
+    def add_listen_socket(self, socktype, sock):
+        info = self.socket_info.get(sock)
+        log("add_listen_socket(%s, %s) info=%s", socktype, sock, info)
+        cleanup = add_listen_socket(socktype, sock, info, self._new_connection, None)
+        if cleanup:
+            self.socket_cleanup.append(cleanup)
+
+    def _new_connection(self, socktype, listener, handle=0):
+        """
+            Accept the new connection,
+            verify that there aren't too many,
+            start a thread to dispatch it to the correct handler.
+        """
+        log("_new_connection%s", (listener, socktype, handle))
+        if self.exit_code is not None:
+            log("ignoring new connection during shutdown")
+            return False
+        try:
+            self.handle_new_connection(socktype, listener, handle)
+        except Exception:
+            log.error("Error handling new connection", exc_info=True)
+        return self.exit_code is None
+
+    def handle_new_connection(self, socktype, listener, handle):
+        socket_info = self.socket_info.get(listener)
+        assert socktype, "cannot find socket type for %s" % listener
+        socket_options = self.socket_options.get(listener, {})
+        assert socktype!="named-pipe"
+        conn = accept_connection(socktype, listener, SOCKET_TIMEOUT, socket_options)
+        if conn is None:
+            return
+        #limit number of concurrent network connections:
+        if len(self._potential_protocols)>=MAX_CONCURRENT_CONNECTIONS:
+            log.error("Error: too many connections (%i)", len(self._potential_protocols))
+            log.error(" ignoring new one: %s", conn.endpoint)
+            conn.close()
+            return
+        sock = conn._socket
+        socktype = conn.socktype
+        peername = conn.endpoint
+        sockname = sock.getsockname()
+        target = peername or sockname
+        log("handle_new_connection%s sockname=%s, target=%s",
+               (conn, socket_info, socket_options), sockname, target)
+        sock.settimeout(SOCKET_TIMEOUT)
+        log_new_connection(conn, socket_info)
+
+        socktype = socktype.lower()
+        protocol = Protocol(self, conn, self.process_network_packet)
+        #protocol.large_packets.append(b"info-response")
+        protocol.socket_type = socktype
+        self._potential_protocols.append(protocol)
+        protocol.authenticators = ()
+        protocol.start()
+        #self.schedule_verify_connection_accepted(protocol, self._accept_timeout)
+
+    def process_network_packet(self, proto, packet):
+        log.info("packet: %s", packet)
+        log.info("packet type: %s", packet[0])
+        if packet[0]==b"hello":
+            caps = typedict(packet[1])
+            proto.parse_remote_caps(caps)
+            proto.enable_compressor_from_caps(caps)
+            proto.enable_encoder_from_caps(caps)
+            hello = get_network_caps()
+            proto.send_now(["hello", hello])
+            info = self.get_info()
+            proto.send_now(["info-response", info])
+            import time
+            time.sleep(1)
+        proto.close()
+
+
     def get_caps(self) -> dict:
         caps = {
             "network-state" : True,