Reproducer for `aiochclient` password leak

aiochclient_password_leak.py

import asyncio
import logging
import socketserver
import time
import threading

import aiochclient
import aiohttp.web


PORT = 1234


def server_thread_target() -> None:
    # start simple server that will drop the connection as soon as possible

    class MyTCPHandler(socketserver.BaseRequestHandler):
        def handle(self):
            pass

    class MyTCPServer(socketserver.TCPServer):
        allow_reuse_address = True

    with MyTCPServer(('localhost', PORT), MyTCPHandler) as server:
        server.serve_forever()


def client_thread_target() -> None:
    async def amain() -> None:
        ch_client = aiochclient.ChClient(
            url=f'http://localhost:{PORT}/',
            database='database',
            user='admin',
            password='secret-password'
        )
        while True:
            try:
                await ch_client.fetchval('SELECT 1')
            except ConnectionResetError:
                continue  # not interesting
            except aiohttp.ServerDisconnectedError:
                continue  # not interesting
            except Exception:
                logging.exception('Leak!')
                exit()

    time.sleep(0.5)  # wait for server to start listen
    asyncio.run(amain())


if __name__ == '__main__':
    threading.Thread(target=server_thread_target, name='Server', daemon=True).start()
    client_thread_target()