from __future__importannotationsimportsysimportcontextlibimportsocketimportsslimporttypingimportthreadingimportosALPN_PROTOCOLS= ["http/1.1"]CERTS_PATH=os.path.join(os.path.dirname(__file__),"certs")DEFAULT_CERTS:dict[str,typing.Any]= {"certfile":os.path.join(CERTS_PATH,"server.crt"),"keyfile":os.path.join(CERTS_PATH,"server.key"),"cert_reqs":ssl.CERT_OPTIONAL,"ca_certs":os.path.join(CERTS_PATH,"cacert.pem"),"alpn_protocols":ALPN_PROTOCOLS,}DEFAULT_CA=os.path.join(CERTS_PATH,"cacert.pem")DEFAULT_CA_KEY=os.path.join(CERTS_PATH,"cacert.key")classSocketServerThread(threading.Thread):"""    :param socket_handler: Callable which receives a socket argument for one        request.    :param ready_event: Event which gets set when the socket handler is        ready to receive requests.    """USE_IPV6=Falsedef__init__(self,socket_handler:typing.Callable[[socket.socket],None],host:str="localhost",ready_event:threading.Event|None=None,    )->None:super().__init__()self.daemon=Trueself.socket_handler=socket_handlerself.host=hostself.ready_event=ready_eventdef_start_server(self)->None:sock=socket.socket(socket.AF_INET)ifsys.platform!="win32":sock.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)withsock:sock.bind((self.host,0))self.port=sock.getsockname()[1]# Once listen() returns, the server socket is readysock.listen(1)ifself.ready_event:self.ready_event.set()self.socket_handler(sock)defrun(self)->None:self._start_server()classNewConnectionError(Exception):passdeforiginal_ssl_wrap_socket(sock:socket.socket,keyfile:StrOrBytesPath|None=None,certfile:StrOrBytesPath|None=None,server_side:bool=False,cert_reqs:ssl.VerifyMode=ssl.CERT_NONE,ssl_version:int=ssl.PROTOCOL_TLS_SERVER,ca_certs:str|None=None,do_handshake_on_connect:bool=True,suppress_ragged_eofs:bool=True,ciphers:str|None=None,)->ssl.SSLSocket:ifserver_sideandnotcertfile:raiseValueError("certfile must be specified for server-side operations")ifkeyfileandnotcertfile:raiseValueError("certfile must be specified")context=ssl.SSLContext(ssl_version)context.verify_mode=cert_reqsifca_certs:context.load_verify_locations(ca_certs)ifcertfile:context.load_cert_chain(certfile,keyfile)ifciphers:context.set_ciphers(ciphers)returncontext.wrap_socket(sock=sock,server_side=server_side,do_handshake_on_connect=do_handshake_on_connect,suppress_ragged_eofs=suppress_ragged_eofs,    )@contextlib.contextmanagerdef_socket_server(handler):ready_event=threading.Event()server_thread=SocketServerThread(socket_handler=handler,ready_event=ready_event,host="localhost"    )server_thread.start()ready_event.wait(5)ifnotready_event.is_set():raiseException("timeout")try:yieldserver_thread.portfinally:server_thread.join()def_test_ssl_failed_fingerprint_verification()->None:defsocket_handler(listener:socket.socket)->None:withlistener.accept()[0]assock:try:ssl_sock=original_ssl_wrap_socket(sock,server_side=True,keyfile=DEFAULT_CERTS["keyfile"],certfile=DEFAULT_CERTS["certfile"],ca_certs=DEFAULT_CA,                )except (ssl.SSLError,ConnectionResetError):passelse:withssl_sock:ssl_sock.send(b"HTTP/1.1 200 OK\r\n"b"Content-Type: text/plain\r\n"b"Content-Length: 5\r\n\r\n"b"Hello"                    )with_socket_server(socket_handler)asport:# GitHub's fingerprint. Valid, but not matching.defrequest()->None:try:try:sock=socket.create_connection(                        ("localhost",port),source_address=None,                    )exceptOSErrorase:raiseNewConnectionError(None,None)ssl.create_default_context().wrap_socket(sock,server_hostname="localhost"                ).close()exceptBaseExceptionase:err=eraisewithcontextlib.suppress(ssl.SSLCertVerificationError):request()# Should not hang, see https://github.com/urllib3/urllib3/issues/529withcontextlib.suppress(NewConnectionError):request()deftest_foo():foriinrange(100):print(i)_test_ssl_failed_fingerprint_verification()deftest_gc():importgcforiinrange(5):gc.collect()defmain():test_foo()test_gc()if__name__=="__main__":sys.exit(main())

python -W error test_socketlevel.py01234Traceback (most recent call last):  File "/usr/lib/python3.12/ssl.py", line 992, in _create    self.getpeername()OSError: [Errno 107] Transport endpoint is not connectedDuring handling of the above exception, another exception occurred:Traceback (most recent call last):  File "/home/graingert/projects/demo-resource-warning/test_socketlevel.py", line 188, in <module>    sys.exit(main())             ^^^^^^  File "/home/graingert/projects/demo-resource-warning/test_socketlevel.py", line 183, in main    test_foo()  File "/home/graingert/projects/demo-resource-warning/test_socketlevel.py", line 172, in test_foo    _test_ssl_failed_fingerprint_verification()  File "/home/graingert/projects/demo-resource-warning/test_socketlevel.py", line 166, in _test_ssl_failed_fingerprint_verification    request()  File "/home/graingert/projects/demo-resource-warning/test_socketlevel.py", line 155, in request    ssl.create_default_context().wrap_socket(  File "/usr/lib/python3.12/ssl.py", line 455, in wrap_socket    return self.sslsocket_class._create(           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^  File "/usr/lib/python3.12/ssl.py", line 1004, in _create    notconn_pre_handshake_data = self.recv(1)                                 ^^^^^^^^^^^^  File "/usr/lib/python3.12/ssl.py", line 1237, in recv    return super().recv(buflen, flags)           ^^^^^^^^^^^^^^^^^^^^^^^^^^^ConnectionResetError: [Errno 104] Connection reset by peerException ignored in: <ssl.SSLSocket fd=5, family=2, type=1, proto=6, laddr=('127.0.0.1', 54864)>ResourceWarning: unclosed <ssl.SSLSocket fd=5, family=2, type=1, proto=6, laddr=('127.0.0.1', 54864)>