Simple TLS sinkhole

simple_tls_sinkhole.py

import asyncio
import ssl
import logging
import socket

logging.basicConfig(filename='tls_logs.log', level=logging.INFO)
#socket.setdefaulttimeout(2)

class TlsSinkholeServer(asyncio.Protocol):
    peername = None

    def connection_made(self, transport):
        self.peername = transport.get_extra_info('peername')
        #print('Connection from {}'.format(self.peername))
        self.transport = transport

    def data_received(self, data):
        message = data.decode()
        text = f'Data received[{self.peername}]: {message}'

        logging.info(text)
        print(text)

        self.transport.close()


async def main():
    sc = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    
    # Create certificates: https://certbot.eff.org/instructions?ws=other&os=ubuntufocal
    sc.load_cert_chain('fullchain.pem', 'privkey.pem')

    loop = asyncio.get_running_loop()

    # coro = asyncio.start_server(handle_connection, '127.0.0.1', port, ssl=sc, loop=loop)

    server = await loop.create_server(
        lambda: TlsSinkholeServer(),
        '0.0.0.0', port=443, ssl=sc)

    async with server:
        await server.serve_forever()


asyncio.run(main())