#474 add option ("tcp-proxy") so non xpra packets can be forwarded to…

… another server, living on a different port

* make all sockets non-blocking (timeout of 0.1s) and only set them to blocking once we have accepted them
* if we find an invalid packet (only for the initial packet), setup a proxy connection to the "tcp-proxy" address
* update to man page and xpra.conf

git-svn-id: https://xpra.org/svn/Xpra/trunk@4948 3bb7dfac-3a0b-4e04-842a-767bc560f471

src/etc/xpra/Xdummy/xpra.conf

@@ -68,6 +68,9 @@ compression_level = 1
 #socket-dir = /tmp
 #socket-dir = ~/.xpra
 
+# Where to send non xpra clients:
+#tcp-proxy = 127.0.0.1:80
+
 # Log file:
 log-file = $DISPLAY.log
 

src/etc/xpra/Xvfb/xpra.conf

@@ -68,6 +68,9 @@ compression_level = 1
 #socket-dir = /tmp
 #socket-dir = ~/.xpra
 
+# Where to send non xpra clients:
+#tcp-proxy = 127.0.0.1:80
+
 # Log file:
 log-file = $DISPLAY.log
 

src/etc/xpra/xpra_Xdummy/xpra.conf

@@ -68,6 +68,9 @@ compression_level = 1
 #socket-dir = /tmp
 #socket-dir = ~/.xpra
 
+# Where to send non xpra clients:
+#tcp-proxy = 127.0.0.1:80
+
 # Log file:
 log-file = $DISPLAY.log
 

src/man/xpra.1

@@ -40,6 +40,7 @@ xpra \- viewer for remote, persistent X applications
 [\fB\-\-clipboard\-filter\-file\fP=\fIFILENAME\fP]
 [\fB\-\-dpi\fP=\fIVALUE\fP]
 [\fB\-\-socket\-dir\fP=\fIDIR\fP]
+[\fB\-\-tcp\-proxy\fP=\fIHOST:PORT\fP]
 .HP
 \fBxpra\fP \fBattach\fP
 [\fI:DISPLAY\fP | \fIssh:[USER@]HOST:DISPLAY\fP | \fItcp:[USER@]HOST:PORT[:DISPLAY]\fP]
@@ -97,6 +98,7 @@ xpra \- viewer for remote, persistent X applications
 [\fB\-\-auth\fP=\fIMODULE\fP]
 [\fB\-\-password\-file\fP=\fIFILENAME\fP]
 [\fB\-\-socket\-dir\fP=\fIDIR\fP]
+[\fB\-\-tcp\-proxy\fP=HOST:PORT\fP]
 .HP
 \fBxpra\fP \fBproxy\fP \fI:DISPLAY\fP
 .HP
@@ -471,6 +473,16 @@ Anyone at all may connect to this port and access your session.
 Use it only if you have special needs, and understand the consequences
 of your actions.
 
+.TP
+\fB\-\-tcp\-proxy\fP=\fIHOST:PORT\fP
+Specifies the address to which non-xpra packets will be forwarded.
+This can be used to share the same TCP port with another
+TCP servers, usually a web server.
+xpra clients will connect as usual, but any client that does not
+speak the xpra protocol will be forwarded to the alternative
+server.
+
+
 .SS Options for start, upgrade and attach
 .TP
 \fB\-\-password\-file\fP=\fIFILENAME\fP

src/xpra/scripts/config.py

@@ -215,6 +215,7 @@ def read_xpra_defaults():
                     "mode"              : str,
                     "window-layout"     : str,
                     "display"           : str,
+                    "tcp-proxy"         : str,
                     #int options:
                     "quality"           : int,
                     "min-quality"       : int,
@@ -302,6 +303,7 @@ def get_defaults():
                     "log-file"          : "$DISPLAY.log",
                     "window-layout"     : "",
                     "display"           : "",
+                    "tcp-proxy"         : "",
                     "quality"           : -1,
                     "min-quality"       : 50,
                     "speed"             : -1,

src/xpra/scripts/main.py

@@ -113,9 +113,14 @@ def parse_cmdline(cmdline):
         group.add_option("--exit-with-children", action="store_true",
                           dest="exit_with_children", default=defaults.exit_with_children,
                           help="Terminate server when --start-child command(s) exit")
+        group.add_option("--tcp-proxy", action="store",
+                          dest="tcp_proxy", default=defaults.tcp_proxy,
+                          metavar="HOST:PORT",
+                          help="The address to which non-xpra packets will be forwarded.")
     else:
         hidden_options["start_child"] = None
         hidden_options["exit_with_children"] = False
+        hidden_options["tcp_proxy"] = ""
     if (supports_server or supports_shadow) and CAN_DAEMONIZE:
         group.add_option("--no-daemon", action="store_false",
                           dest="daemon", default=True,

src/xpra/server/proxy_server.py

@@ -20,7 +20,6 @@
 from xpra.scripts.main import parse_display_name, connect_to
 from xpra.scripts.server import deadly_signal
 from xpra.net.protocol import Protocol, Compressed, compressed_wrapper, new_cipher_caps, get_network_caps
-from xpra.net.bytestreams import SocketConnection
 from xpra.os_util import Queue, SIGNAMES
 from xpra.util import typedict
 from xpra.daemon_thread import make_daemon_thread
@@ -192,9 +191,7 @@ def do_start_proxy():
                     log.error("IO threads have failed to terminate!")
                     return
                 #now we can go back to using blocking sockets:
-                #FIXME: this is a bit ugly, but less intrusive than the alternative?
-                if isinstance(client_conn, SocketConnection):
-                    client_conn._socket.settimeout(None)
+                self.set_socket_timeout(client_conn, None)
                 client_conn.set_active(True)
 
                 assert uid!=0 and gid!=0

src/xpra/server/server_base.py

@@ -405,11 +405,8 @@ def cleanup_source(self, protocol):
             self._potential_protocols.remove(protocol)
         return source
 
-    def verify_connection_accepted(self, protocol):
-        if not protocol._closed and protocol in self._potential_protocols and protocol not in self._server_sources:
-            log.error("connection timedout: %s", protocol)
-            self.send_disconnect(protocol, "login timeout")
-
+    def is_timedout(self, protocol):
+        return ServerCore.is_timedout(self, protocol) and protocol not in self._server_sources
 
     def no_more_clients(self):
         #so it is now safe to clear them:
@@ -472,6 +469,9 @@ def hello_oked(self, proto, packet, c, auth_caps):
         #max packet size from client (the biggest we can get are clipboard packets)
         proto.max_packet_size = 1024*1024  #1MB
         proto.aliases = c.dictget("aliases")
+        #use blocking sockets from now on:
+        self.set_socket_timeout(proto._conn, None)
+
         def drop_client(reason="unknown"):
             self.disconnect_client(proto, reason)
         def get_window_id(wid):

src/xpra/server/server_core.py

@@ -20,17 +20,20 @@
 log = Logger()
 
 import xpra
-from xpra.scripts.main import SOCKET_TIMEOUT
+from xpra.scripts.main import SOCKET_TIMEOUT, _socket_connect
 from xpra.scripts.config import ENCRYPTION_CIPHERS
 from xpra.scripts.server import deadly_signal
 from xpra.net.bytestreams import SocketConnection
 from xpra.os_util import set_application_name, load_binary_file, get_machine_id, get_user_uuid, SIGNAMES
 from xpra.version_util import version_compat_check, add_version_info, get_platform_info
-from xpra.net.protocol import Protocol, use_lz4, use_rencode, new_cipher_caps, get_network_caps
+from xpra.net.protocol import Protocol, use_lz4, use_rencode, new_cipher_caps, get_network_caps, repr_ellipsized
 from xpra.server.background_worker import stop_worker
+from xpra.daemon_thread import make_daemon_thread
+from xpra.server.proxy import XpraProxy
 from xpra.util import typedict
 
 
+
 MAX_CONCURRENT_CONNECTIONS = 20
 
 
@@ -99,11 +102,13 @@ def __init__(self):
         self._upgrading = False
         #networking bits:
         self._potential_protocols = []
+        self._tcp_proxy_clients = []
+        self._tcp_proxy = ""
         self._aliases = {}
         self._reverse_aliases = {}
         self.socket_types = {}
         self._max_connections = MAX_CONCURRENT_CONNECTIONS
-        self._socket_timeout = None
+        self._socket_timeout = 0.1
 
         self.session_name = "Xpra"
 
@@ -130,6 +135,7 @@ def init(self, opts):
         self.session_name = opts.session_name
         set_application_name(self.session_name)
 
+        self._tcp_proxy = opts.tcp_proxy
         self.encryption_keyfile = opts.encryption_keyfile
         self.password_file = opts.password_file
         self.compression_level = opts.compression_level
@@ -259,8 +265,11 @@ def do_run(self):
         raise NotImplementedError()
 
     def cleanup(self, *args):
+        log("cleanup() stopping %s tcp proxy clients: %s", len(self._tcp_proxy_clients), self._tcp_proxy_clients)
+        for p in list(self._tcp_proxy_clients):
+            p.quit()
         log("cleanup will disconnect: %s", self._potential_protocols)
-        for proto in self._potential_protocols:
+        for proto in list(self._potential_protocols):
             if self._upgrading:
                 reason = "upgrading/exiting"
             else:
@@ -291,13 +300,60 @@ def _new_connection(self, listener, *args):
         protocol = Protocol(self, sc, self.process_packet)
         protocol.large_packets.append("info-response")
         protocol.authenticator = None
+        protocol.invalid_header = self.invalid_header
         self._potential_protocols.append(protocol)
         protocol.start()
         self.timeout_add(SOCKET_TIMEOUT*1000, self.verify_connection_accepted, protocol)
         return True
 
+    def invalid_header(self, proto, data):
+        log("invalid_header(%s, %s)", proto, repr_ellipsized(data))
+        if proto.input_packetcount==0 and self._tcp_proxy:
+            #look for http get:
+            if data[:4]=="GET ":
+                self.start_tcp_proxy(proto, data)
+                return
+        err = "invalid packet header byte: '%s', not an xpra client?" % hex(ord(data[0]))
+        if len(data)>1:
+            err += " read buffer=0x%s" % repr_ellipsized(data)
+        proto.gibberish(err, data)
+
+    def start_tcp_proxy(self, proto, data):
+        log("start_tcp_proxy(%s, %s)", proto, data[:10])
+        client_connection = proto.steal_connection()
+        self._potential_protocols.remove(proto)
+        #connect to web server:
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.settimeout(10)
+        host, port = self._tcp_proxy.split(":", 1)
+        try:
+            web_server_connection = _socket_connect(sock, (host, int(port)), "web-proxy-for-%s" % proto, "tcp")
+        except:
+            log.warn("failed to connect to proxy: %s:%s", host, port)
+            proto.gibberish("invalid packet header", data)
+            return
+        log("proxy connected to tcp server at %s:%s : %s", host, port, web_server_connection)
+        web_server_connection.write(data)
+        p = XpraProxy(client_connection, web_server_connection)
+        self._tcp_proxy_clients.append(p)
+        def run_proxy():
+            p.run()
+            log("run_proxy() %s ended", p)
+            if p in self._tcp_proxy_clients:
+                self._tcp_proxy_clients.remove(p)
+        t = make_daemon_thread(run_proxy, "web-proxy-for-%s" % proto)
+        t.start()
+        log.info("client %s forwarded to proxy server %s:%s", client_connection, host, port)
+
+    def is_timedout(self, protocol):
+        #subclasses may override this method (ServerBase does)
+        return not protocol._closed and protocol in self._potential_protocols and \
+            protocol not in self._tcp_proxy_clients
+
     def verify_connection_accepted(self, protocol):
-        raise NotImplementedError()
+        if self.is_timedout(protocol):
+            log.error("connection timedout: %s", protocol)
+            self.send_disconnect(protocol, "login timeout")
 
     def send_disconnect(self, proto, reason):
         log("send_disconnect(%s, %s)", proto, reason)
@@ -368,6 +424,11 @@ def _process_hello(self, proto, packet):
                 log.error("server error processing new connection from %s", proto, exc_info=True)
                 self.disconnect_client(proto, "server error accepting new connection")
 
+    def set_socket_timeout(self, conn, timeout=None):
+        #FIXME: this is ugly, but less intrusive than the alternative?
+        if isinstance(conn, SocketConnection):
+            conn._socket.settimeout(timeout)
+
 
     def verify_hello(self, proto, c):
         remote_version = c.strget("version")