def _process_gibberish(self, packet):
     (_, message, data) = packet
     p = self._protocol
     show_as_text = p and p.input_packetcount==0 and all(c in string.printable for c in bytestostr(data))
     if show_as_text:
         #looks like the first packet back is just text, print it:
         data = bytestostr(data)
         if data.find("Traceback "):
             for x in data.split("\n"):
                 log.warn(x.strip("\r"))
         else:
             log.warn("Failed to connect, received: %s", repr_ellipsized(data.strip("\n").strip("\r")))
     else:
         log.warn("Received uninterpretable nonsense: %s", message)
         log.warn(" packet no %i data: %s", p.input_packetcount, repr_ellipsized(data))
     if str(data).find("assword")>0:
         self.warn_and_quit(EXIT_SSH_FAILURE,
                           "Your ssh program appears to be asking for a password."
                          + GOT_PASSWORD_PROMPT_SUGGESTION)
     elif str(data).find("login")>=0:
         self.warn_and_quit(EXIT_SSH_FAILURE,
                          "Your ssh program appears to be asking for a username.\n"
                          "Perhaps try using something like 'ssh:[email protected]:display'?")
     else:
         self.quit(EXIT_PACKET_FAILURE)

    def start_tcp_proxy(self, proto, data):
        proxylog("start_tcp_proxy(%s, '%s')", proto, repr_ellipsized(data))
        try:
            self._potential_protocols.remove(proto)
        except:
            pass  # might already have been removed by now
        proxylog("start_tcp_proxy: protocol state before stealing: %s", proto.get_info(alias_info=False))
        # any buffers read after we steal the connection will be placed in this temporary queue:
        temp_read_buffer = Queue()
        client_connection = proto.steal_connection(temp_read_buffer.put)
        # 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:
            proxylog.warn("failed to connect to proxy: %s:%s", host, port)
            proto.gibberish("invalid packet header", data)
            return
        proxylog("proxy connected to tcp server at %s:%s : %s", host, port, web_server_connection)
        sock.settimeout(self._socket_timeout)

        ioe = proto.wait_for_io_threads_exit(0.5 + self._socket_timeout)
        if not ioe:
            proxylog.warn("proxy failed to stop all existing network threads!")
            self.disconnect_protocol(proto, "internal threading error")
            return
        # now that we own it, we can start it again:
        client_connection.set_active(True)
        # and we can use blocking sockets:
        self.set_socket_timeout(client_connection, None)
        # prevent deadlocks on exit:
        sock.settimeout(1)

        proxylog("pushing initial buffer to its new destination: %s", repr_ellipsized(data))
        web_server_connection.write(data)
        while not temp_read_buffer.empty():
            buf = temp_read_buffer.get()
            if buf:
                proxylog("pushing read buffer to its new destination: %s", repr_ellipsized(buf))
                web_server_connection.write(buf)
        p = XpraProxy(client_connection.target, client_connection, web_server_connection)
        self._tcp_proxy_clients.append(p)
        proxylog.info(
            "client connection from %s forwarded to proxy server on %s:%s", client_connection.target, host, port
        )
        p.run()
        proxylog("run_proxy() %s ended", p)
        if p in self._tcp_proxy_clients:
            self._tcp_proxy_clients.remove(p)

 def invalid_header(self, proto, data):
     log("invalid_header(%s, %s)", proto, repr_ellipsized(data))
     if proto.input_packetcount==0 and self._tcp_proxy:
         self.start_tcp_proxy(proto, data)
         return
     err = "invalid packet format, not an xpra client?"
     proto.gibberish(err, data)

 def _process_gibberish(self, packet):
     (_, message, data) = packet
     p = self._protocol
     show_as_text = p and p.input_packetcount==0 and all(c in string.printable for c in bytestostr(data))
     if show_as_text:
         #looks like the first packet back is just text, print it:
         data = bytestostr(data)
         if data.find("Traceback "):
             for x in data.split("\n"):
                 netlog.warn(x.strip("\r"))
         else:
             netlog.warn("Failed to connect, received: %s", repr_ellipsized(data.strip("\n").strip("\r")))
     else:
         netlog.warn("Received uninterpretable nonsense: %s", message)
         netlog.warn(" packet no %i data: %s", p.input_packetcount, repr_ellipsized(data))
     self.quit(EXIT_PACKET_FAILURE)

	def test_ssl_socket(self):
		server = None
		display_no = self.find_free_display_no()
		display = ":%s" % display_no
		tcp_port = get_free_tcp_port()
		ssl_port = get_free_tcp_port()
		try:
			tmpdir = tempfile.mkdtemp(suffix='ssl-xpra')
			certfile = os.path.join(tmpdir, "self.pem")
			openssl_command = [
								"openssl", "req", "-new", "-newkey", "rsa:4096", "-days", "2", "-nodes", "-x509",
								"-subj", "/C=US/ST=Denial/L=Springfield/O=Dis/CN=localhost",
    							"-keyout", certfile, "-out", certfile,
    							]
			openssl = self.run_command(openssl_command)
			assert pollwait(openssl, 10)==0, "openssl certificate generation failed"
			cert_data = load_binary_file(certfile)
			log("generated cert data: %s", repr_ellipsized(cert_data))
			if not cert_data:
				#cannot run openssl? (happens from rpmbuild)
				log.warn("SSL test skipped, cannot run '%s'", b" ".join(openssl_command))
				return
			server_args = [
							"--bind-tcp=0.0.0.0:%i" % tcp_port,
							"--bind-ssl=0.0.0.0:%i" % ssl_port,
							"--ssl=on",
							"--ssl-cert=%s" % certfile]

			log("starting test ssl server on %s", display)
			server = self.start_server(display, *server_args)

			#test it with openssl client:
			for port in (tcp_port, ssl_port):
				openssl_verify_command = "openssl s_client -connect 127.0.0.1:%i -CAfile %s < /dev/null" % (port, certfile)
				openssl = self.run_command(openssl_verify_command, shell=True)
				assert pollwait(openssl, 10)==0, "openssl certificate verification failed"

			def test_connect(uri, exit_code, *client_args):
				cmd = ["info", uri] + list(client_args)
				client = self.run_xpra(cmd)
				r = pollwait(client, 5)
				if client.poll() is None:
					client.terminate()
				assert r==exit_code, "expected info client to return %s but got %s" % (exit_code, client.poll())
			noverify = "--ssl-server-verify-mode=none"
			#connect to ssl socket:
			test_connect("ssl/127.0.0.1:%i/" % ssl_port, EXIT_OK, noverify)
			#tcp socket should upgrade:
			test_connect("ssl/127.0.0.1:%i/" % tcp_port, EXIT_OK, noverify)
			#self signed cert should fail without noverify:
			test_connect("ssl/127.0.0.1:%i/" % ssl_port, EXIT_CONNECTION_LOST)
			test_connect("ssl/127.0.0.1:%i/" % tcp_port, EXIT_CONNECTION_LOST)

		finally:
			shutil.rmtree(tmpdir)
			if server:
				server.terminate()

 def _copy_loop(self, log_name, from_conn, to_conn):
     #log("XpraProxy._copy_loop(%s, %s, %s)", log_name, from_conn, to_conn)
     try:
         while not self._closed:
             log("%s: waiting for data", log_name)
             buf = untilConcludes(self.is_active, from_conn.read, PROXY_BUFFER_SIZE)
             if not buf:
                 log("%s: connection lost", log_name)
                 return
             if SHOW_DATA:
                 log("%s: %s bytes: %s", log_name, len(buf), repr_ellipsized(buf))
                 log("%s:           %s", log_name, repr_ellipsized(binascii.hexlify(buf)))
             while buf and not self._closed:
                 log("%s: writing %s bytes", log_name, len(buf))
                 written = untilConcludes(self.is_active, to_conn.write, buf)
                 buf = buf[written:]
                 log("%s: written %s bytes", log_name, written)
     except Exception as e:
         log("%s: %s", log_name, e)
         self.quit()

 def __init__(self, uid, gid, env_options, session_options, socket_dir,
              video_encoder_modules, csc_modules,
              client_conn, client_state, cipher, encryption_key, server_conn, caps, message_queue):
     Process.__init__(self, name=str(client_conn))
     self.uid = uid
     self.gid = gid
     self.env_options = env_options
     self.session_options = session_options
     self.socket_dir = socket_dir
     self.video_encoder_modules = video_encoder_modules
     self.csc_modules = csc_modules
     self.client_conn = client_conn
     self.client_state = client_state
     self.cipher = cipher
     self.encryption_key = encryption_key
     self.server_conn = server_conn
     self.caps = caps
     log("ProxyProcess%s", (uid, gid, env_options, session_options, socket_dir,
                            video_encoder_modules, csc_modules,
                            client_conn, repr_ellipsized(str(client_state)), cipher, encryption_key, server_conn,
                            "%s: %s.." % (type(caps), repr_ellipsized(str(caps))), message_queue))
     self.client_protocol = None
     self.server_protocol = None
     self.exit = False
     self.main_queue = None
     self.message_queue = message_queue
     self.encode_queue = None            #holds draw packets to encode
     self.encode_thread = None
     self.video_encoding_defs = None
     self.video_encoders = None
     self.video_encoders_last_used_time = None
     self.video_encoder_types = None
     self.video_helper = None
     self.lost_windows = None
     #for handling the local unix domain socket:
     self.control_socket_cleanup = None
     self.control_socket = None
     self.control_socket_thread = None
     self.control_socket_path = None
     self.potential_protocols = []
     self.max_connections = MAX_CONCURRENT_CONNECTIONS

 def process_packet(self, proto, packet):
     command = bytestostr(packet[0])
     if command==Protocol.CONNECTION_LOST:
         log("connection-lost: %s, calling stop", packet[1:])
         self.net_stop()
         return
     elif command==Protocol.GIBBERISH:
         log.warn("gibberish received:")
         log.warn(" %s", repr_ellipsized(packet[1], limit=80))
         log.warn(" stopping")
         self.net_stop()
         return
     elif command=="stop":
         log("received stop message")
         self.net_stop()
         return
     elif command=="exit":
         log("received exit message")
         sys.exit(0)
         return
     #make it easier to hookup signals to methods:
     attr = command.replace("-", "_")
     if self.method_whitelist is not None and attr not in self.method_whitelist:
         log.warn("invalid command: %s (not in whitelist: %s)", attr, self.method_whitelist)
         return
     wo = self.wrapped_object
     if not wo:
         log("wrapped object is no more, ignoring method call '%s'", attr)
         return
     method = getattr(wo, attr, None)
     if not method:
         log.warn("unknown command: '%s'", attr)
         log.warn(" packet: '%s'", repr_ellipsized(str(packet)))
         return
     if DEBUG_WRAPPER:
         log("calling %s.%s%s", wo, attr, str(tuple(packet[1:]))[:128])
     self.idle_add(method, *packet[1:])
     INJECT_FAULT(proto)

 def check_packet_size(size_to_check, packet_header):
     if self._closed:
         return False
     log(
         "check_packet_size(%s, 0x%s) limit is %s",
         size_to_check,
         repr_ellipsized(packet_header),
         self.max_packet_size,
     )
     if size_to_check > self.max_packet_size:
         msg = "packet size requested is %s but maximum allowed is %s" % (
             size_to_check,
             self.max_packet_size,
         )
         self.invalid(msg, packet_header)
     return False

 def t(s, ev, remainder=""):
     try:
         rv, rr = self.decode(s)
         #print("decode(%s)=%s (%s)" % (s, rv, type(rv)))
         _cmp(rv, ev)
     except Exception as e:
         print("error on decoding of '%s'" % repr_ellipsized(s))
         raise e
     rrstr = s[rr:]
     assert rrstr == remainder, "expected remainder value '%s' but got %s" % (remainder, rrstr)
     # With gibberish added:
     g_str = s + "asdf"
     rv, rr = self.decode(g_str)
     _cmp(rv, ev)
     rrstr = g_str[rr:]
     assert rrstr.endswith("asdf")

    def invalid_header(self, proto, data):
        netlog(
            "invalid_header(%s, %s bytes: '%s') input_packetcount=%s, tcp_proxy=%s",
            proto,
            len(data or ""),
            repr_ellipsized(data),
            proto.input_packetcount,
            self._tcp_proxy,
        )
        if proto.input_packetcount == 0 and self._tcp_proxy and not proto._closed:
            # start a new proxy in a thread
            def run_proxy():
                self.start_tcp_proxy(proto, data)

            t = make_daemon_thread(run_proxy, "web-proxy-for-%s" % proto)
            t.start()
            return
        err = "invalid packet format, not an xpra client?"
        proto.gibberish(err, data)

 def invalid_header(self, proto, data):
     err = "invalid packet header: '%s'" % binascii.hexlify(data[:8])
     if len(data) > 1:
         err += " read buffer=%s" % repr_ellipsized(data)
     self.gibberish(err, data)

    def do_read_parse_thread_loop(self):
        """
            Process the individual network packets placed in _read_queue.
            Concatenate the raw packet data, then try to parse it.
            Extract the individual packets from the potentially large buffer,
            saving the rest of the buffer for later, and optionally decompress this data
            and re-construct the one python-object-packet from potentially multiple packets (see packet_index).
            The 8 bytes packet header gives us information on the packet index, packet size and compression.
            The actual processing of the packet is done via the callback process_packet_cb,
            this will be called from this parsing thread so any calls that need to be made
            from the UI thread will need to use a callback (usually via 'idle_add')
        """
        read_buffer = None
        payload_size = -1
        padding_size = 0
        packet_index = 0
        compression_level = False
        packet = None
        raw_packets = {}
        while not self._closed:
            buf = self._read_queue.get()
            if not buf:
                log("read thread: empty marker, exiting")
                self.idle_add(self.close)
                return
            if read_buffer:
                read_buffer = read_buffer + buf
            else:
                read_buffer = buf
            bl = len(read_buffer)
            while not self._closed:
                packet = None
                bl = len(read_buffer)
                if bl<=0:
                    break
                if payload_size<0:
                    if read_buffer[0] not in ("P", ord("P")):
                        self._invalid_header(read_buffer)
                        return
                    if bl<8:
                        break   #packet still too small
                    #packet format: struct.pack('cBBBL', ...) - 8 bytes
                    _, protocol_flags, compression_level, packet_index, data_size = unpack_header(read_buffer[:8])

                    #sanity check size (will often fail if not an xpra client):
                    if data_size>self.abs_max_packet_size:
                        self._invalid_header(read_buffer)
                        return

                    bl = len(read_buffer)-8
                    if protocol_flags & FLAGS_CIPHER:
                        if self.cipher_in_block_size==0 or not self.cipher_in_name:
                            cryptolog.warn("received cipher block but we don't have a cipher to decrypt it with, not an xpra client?")
                            self._invalid_header(read_buffer)
                            return
                        padding_size = self.cipher_in_block_size - (data_size % self.cipher_in_block_size)
                        payload_size = data_size + padding_size
                    else:
                        #no cipher, no padding:
                        padding_size = 0
                        payload_size = data_size
                    assert payload_size>0, "invalid payload size: %i" % payload_size
                    read_buffer = read_buffer[8:]

                    if payload_size>self.max_packet_size:
                        #this packet is seemingly too big, but check again from the main UI thread
                        #this gives 'set_max_packet_size' a chance to run from "hello"
                        def check_packet_size(size_to_check, packet_header):
                            if self._closed:
                                return False
                            log("check_packet_size(%s, 0x%s) limit is %s", size_to_check, repr_ellipsized(packet_header), self.max_packet_size)
                            if size_to_check>self.max_packet_size:
                                msg = "packet size requested is %s but maximum allowed is %s" % \
                                              (size_to_check, self.max_packet_size)
                                self.invalid(msg, packet_header)
                            return False
                        self.timeout_add(1000, check_packet_size, payload_size, read_buffer[:32])

                if bl<payload_size:
                    # incomplete packet, wait for the rest to arrive
                    break

                #chop this packet from the buffer:
                if len(read_buffer)==payload_size:
                    raw_string = read_buffer
                    read_buffer = ''
                else:
                    raw_string = read_buffer[:payload_size]
                    read_buffer = read_buffer[payload_size:]
                #decrypt if needed:
                data = raw_string
                if self.cipher_in and protocol_flags & FLAGS_CIPHER:
                    cryptolog("received %i %s encrypted bytes with %s padding", payload_size, self.cipher_in_name, padding_size)
                    data = self.cipher_in.decrypt(raw_string)
                    if padding_size > 0:
                        def debug_str(s):
                            try:
                                return binascii.hexlify(bytearray(s))
                            except:
                                return csv(list(str(s)))
#.........这里部分代码省略.........

 def unexpected_packet(packet):
     if packet:
         log.warn("Warning: received an unexpected packet on the proxy connection %s:", client_proto)
         log.warn(" %s", repr_ellipsized(packet))

 def unexpected_packet(packet):
     if packet:
         log.warn("received an unexpected packet on the proxy connection: %s", repr_ellipsized(packet))

 def _process_gibberish(self, proto, packet):
     (_, message, data) = packet
     netlog("Received uninterpretable nonsense from %s: %s", proto, message)
     netlog(" data: %s", repr_ellipsized(data))
     self.disconnect_client(proto, message)

 def _process_invalid(self, protocol, packet):
     (_, message, data) = packet
     netlog("Received invalid packet: %s", message)
     netlog(" data: %s", repr_ellipsized(data))
     self.disconnect_client(protocol, message)

 def gibberish(self, *args):
     log.warn("%s stopping on gibberish:", self.description)
     log.warn(" %s", repr_ellipsized(args[1], limit=80))
     self.stop()

 def encode(self, packet_in):
     """
     Given a packet (tuple or list of items), converts it for the wire.
     This method returns all the binary packets to send, as an array of:
     (index, compression_level and compression flags, binary_data)
     The index, if positive indicates the item to populate in the packet
     whose index is zero.
     ie: ["blah", [large binary data], "hello", 200]
     may get converted to:
     [
         (1, compression_level, [large binary data now zlib compressed]),
         (0,                 0, bencoded/rencoded(["blah", '', "hello", 200]))
     ]
     """
     packets = []
     packet = list(packet_in)
     level = self.compression_level
     size_check = LARGE_PACKET_SIZE
     min_comp_size = 378
     for i in range(1, len(packet)):
         item = packet[i]
         ti = type(item)
         if ti in (int, long, bool, dict, list, tuple):
             continue
         l = len(item)
         if ti == Uncompressed:
             # this is a marker used to tell us we should compress it now
             # (used by the client for clipboard data)
             item = item.compress()
             packet[i] = item
             ti = type(item)
             # (it may now be a "Compressed" item and be processed further)
         if ti in (Compressed, LevelCompressed):
             # already compressed data (usually pixels, cursors, etc)
             if not item.can_inline or l > INLINE_SIZE:
                 il = 0
                 if ti == LevelCompressed:
                     # unlike Compressed (usually pixels, decompressed in the paint thread),
                     # LevelCompressed is decompressed by the network layer
                     # so we must tell it how to do that and pass the level flag
                     il = item.level
                 packets.append((i, il, item.data))
                 packet[i] = ""
             else:
                 # data is small enough, inline it:
                 packet[i] = item.data
                 min_comp_size += l
                 size_check += l
         elif ti in (str, bytes) and level > 0 and l > LARGE_PACKET_SIZE:
             log.warn(
                 "found a large uncompressed item in packet '%s' at position %s: %s bytes", packet[0], i, len(item)
             )
             # add new binary packet with large item:
             cl, cdata = self._compress(item, level)
             packets.append((i, cl, cdata))
             # replace this item with an empty string placeholder:
             packet[i] = ""
         elif ti not in (str, bytes):
             log.warn("unexpected data type %s in %s packet: %s", ti, packet[0], repr_ellipsized(item))
     # now the main packet (or what is left of it):
     packet_type = packet[0]
     self.output_stats[packet_type] = self.output_stats.get(packet_type, 0) + 1
     if USE_ALIASES and self.send_aliases and packet_type in self.send_aliases:
         # replace the packet type with the alias:
         packet[0] = self.send_aliases[packet_type]
     try:
         main_packet, proto_version = self._encoder(packet)
     except Exception as e:
         if self._closed:
             return [], 0
         log.error("failed to encode packet: %s", packet, exc_info=True)
         # make the error a bit nicer to parse: undo aliases:
         packet[0] = packet_type
         self.verify_packet(packet)
         raise e
     if len(main_packet) > size_check and packet_in[0] not in self.large_packets:
         log.warn(
             "found large packet (%s bytes): %s, argument types:%s, sizes: %s, packet head=%s",
             len(main_packet),
             packet_in[0],
             [type(x) for x in packet[1:]],
             [len(str(x)) for x in packet[1:]],
             repr_ellipsized(packet),
         )
     # compress, but don't bother for small packets:
     if level > 0 and len(main_packet) > min_comp_size:
         cl, cdata = self._compress(main_packet, level)
         packets.append((0, cl, cdata))
     else:
         packets.append((0, 0, main_packet))
     return packets, proto_version

 def _process_invalid(self, packet):
     (_, message, data) = packet
     log.info("Received invalid packet: %s", message)
     log(" data: %s", repr_ellipsized(data))
     self.quit(EXIT_PACKET_FAILURE)