import asyncio import contextlib import logging import socket import ssl import subprocess import tempfile @contextlib.contextmanager def server_ssl_context(host): with tempfile.NamedTemporaryFile() as keyfile, tempfile.NamedTemporaryFile() as certfile: subprocess.run([ 'openssl', 'req', '-new', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:prime256v1', '-keyout', keyfile.name, '-nodes', '-x509', '-days', '365', '-subj', f'/CN={host}', '-out', certfile.name, ], check=True, shell=False) context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) context.load_cert_chain(certfile.name, keyfile.name) yield context @contextlib.contextmanager def client_ssl_context(): try: context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH) context.check_hostname = False context.verify_mode = ssl.CERT_NONE yield context finally: pass async def client_handle(reader, writer): ... async def main(host, port): with server_ssl_context(host) as server_context, client_ssl_context() as client_context: await asyncio.start_server(client_handle, host, port, ssl=server_context) reader, writer = await asyncio.open_connection(host, port, ssl=client_context) transport = writer._transport from asyncio.sslproto import _SSLProtocolTransport assert isinstance(transport, _SSLProtocolTransport) inner_transport = transport._ssl_protocol._transport from asyncio.selector_events import _SelectorSocketTransport assert isinstance(inner_transport, _SelectorSocketTransport) sock: socket.socket = inner_transport._sock assert isinstance(sock, socket.socket) # Simulate a broken pipe, this invokes the OS error "Bad file descriptor" # but triggers the same flow as "Broken Pipe" sock.close() # Invoke write so socket returns an OSError print('[Client] Sending x6: %r' % 'Hello, world!') # Increment _conn_lost to more than 5 to trigger the logging. # This silently fails, but the logging is triggered. for i in range(6): writer.write('Hello, world!'.encode()) await writer.drain() print(f"{inner_transport._conn_lost=}, {transport.is_closing()=}, {inner_transport.is_closing()=}") if __name__ == '__main__': logging.basicConfig(level=logging.DEBUG) asyncio.run(main('localhost', 8443), debug=True)