Python ssl 模块，PROTOCOL_TLSv1_1() 实例源码

def get_ssl_version(sock):
        # Seth behaves differently depeding on the TLS protocol
        # https://bugs.python.org/issue31453
        # This is an ugly hack (as if the rest of this wasn't...)
    versions = [
        ssl.PROTOCOL_TLSv1,
        ssl.PROTOCOL_TLSv1_1,
        ssl.PROTOCOL_TLSv1_2,
        ]
    firstbytes = sock.recv(16, socket.MSG_PEEK)
    try:
        return versions[firstbytes[10]-1]
    except IndexError:
        print("Unexpected SSL version: %s" % hexlify(firstbytes))
        return versions[-1]


#  def launch_rdp_client():
#      time.sleep(1)
#      p = subprocess.Popen(
#          ["xfreerdp",
#           "/v:%s:%d" % (args.bind_ip, consts.RELAY_PORT),
#           "/u:%s\\%s" % (domain, user),
#          ],
#      )

def run(self):
        global httpd
        print('starting server...')
        server_address = (self.ip,self.port)
        try:
            httpd = ThreadedHTTPServer(server_address, testHTTPServer_RequestHandler)
            if self.https_allow and self.cert_file:
                if os.path.exists(self.cert_file):
                    httpd.socket = ssl.wrap_socket(
                        httpd.socket,certfile=self.cert_file,
                        ssl_version=ssl.PROTOCOL_TLSv1_1)
        except:
            txt = 'Your local IP changed..or port is blocked\n..Trying to find new IP'
            #subprocess.Popen(['notify-send',txt])
            send_notification(txt)
            self.ip = get_ip()
            txt = 'Your New Address is '+self.ip + '\n Please restart the player'
            #subprocess.Popen(['notify-send',txt])
            send_notification(txt)
            change_config_file(self.ip,self.port)
            server_address = (self.ip,self.port)
            httpd = ThreadedHTTPServer(server_address, testHTTPServer_RequestHandler)
        print('running server...at..'+self.ip+':'+str(self.port))
        httpd.serve_forever()

def run(self):
        global httpd
        print('starting server...')
        server_address = (self.ip,self.port)
        try:
            httpd = ThreadedHTTPServer(server_address, testHTTPServer_RequestHandler)
            if self.https_allow and self.cert_file:
                if os.path.exists(self.cert_file):
                    httpd.socket = ssl.wrap_socket(
                        httpd.socket,certfile=self.cert_file,
                        ssl_version=ssl.PROTOCOL_TLSv1_1)
        except:
            txt = 'Your local IP changed..or port is blocked\n..Trying to find new IP'
            #subprocess.Popen(['notify-send',txt])
            send_notification(txt)
            self.ip = get_ip()
            txt = 'Your New Address is '+self.ip + '\n Please restart the player'
            #subprocess.Popen(['notify-send',txt])
            send_notification(txt)
            change_config_file(self.ip,self.port)
            server_address = (self.ip,self.port)
            httpd = ThreadedHTTPServer(server_address, testHTTPServer_RequestHandler)
        print('running server...at..'+self.ip+':'+str(self.port))
        httpd.serve_forever()

def test_tls1_1_disabled(self):
        self.assertRaises(socket.error,
                          self._connect_socket,
                          ssl_version=ssl.PROTOCOL_TLSv1_1)

def main(self):
        # Check Python version
        py_ver = sys.version_info
        if (
            py_ver.major < 2
            or (
                py_ver.major == 2
                and (
                    py_ver.minor < 7
                    or (py_ver.minor >= 7 and py_ver.micro < 10)
                )
            )
        ):
            raise Exception('Your version of Python and Python-ssl are too old. Please upgrade to more "current" versions')

        # Set up SSL/TLS context
        tls_version_table = {
            'SSLv3': ssl.PROTOCOL_SSLv23,
            'TLSv1': ssl.PROTOCOL_TLSv1,
            'TLSv1.1': ssl.PROTOCOL_TLSv1_1,
            'TLSv1.2': ssl.PROTOCOL_TLSv1_2,
        }

        tls_version = tls_version_table[self.version]

        ctx = ssl.SSLContext(tls_version)
        if not isinstance(self.alpn, type(None)):
            ctx.set_alpn_protocols(','.join(self.alpn))
        ctx.set_ciphers(self.cipher_suites)
        if not isinstance(self.cacert_file, type(None)):
            ctx.load_verify_locations(cafile=self.cacert_file)

        ctx.load_cert_chain(self.cert_file, self.key_file)

        if self.protocol == 'IPv4':
            server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        else:
            server_sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)

        server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)

        ssl_sock = ctx.wrap_socket(server_sock, server_side=True)

        ssl_sock.bind(('' if isinstance(self.ip_dst, type(None)) else self.ip_dst, self.port_dst))

        ssl_sock.listen(self.backlog_size)
        ssl_sock.settimeout(self.timeout)

        self._serve(ssl_sock)

        try:
            server_sock = ssl_sock.unwrap()
            server_sock.shutdown(socket.SHUT_RDWR)
        except:
            pass
        finally:
            server_sock.close()