asyncio Man-in-the-Middle Proxy – allowing for bidirectional streaming

proxy.py

"""
Bidirectional asyncio TCP man-in-the-middle proxy
allowing for both serverside- and clientside-triggered
network events.

github/alfonsrv, 2022
Licensed under Mozilla Public License 2.0
"""

import asyncio

REMOTE_SERVER = '10.100.30.199'
SERVER_BIND = '127.0.0.1'
SERVER_PORT = 7001

class AsyncioProxy:
    client_writer = None
    server_writer = None
    client_peer = None  # tuple

    async def _receive(self, reader, prefix: str) -> Tuple[bytes, int]:
        data = await reader.read(100)
        print(f'[{prefix}] Received: {data}')
        return data.decode('utf-8')

    async def _receive_client(self, reader):
        while True:
            data = await self._receive(reader=reader, prefix='C')
            await self._send_server(message=data)

    async def _receive_server(self, reader):
        while True:
            data = await self._receive(reader=reader, prefix='S')
            await self._send_client(data)

    async def _send_client(self, message: str):
        """ Sends/forwards data from server to client """
        print(f'Sending to Client: {message}')
        self.client_writer.write(message)

    async def _send_server(self, message: str):
        """ Sends/forwards data to server """
        print(f'Sending to Server: {message}')
        self.server_writer.write(message)

    async def _handle_client(self, server_reader, server_writer):
        self.client_peer = server_writer.get_extra_info("peername")
        print(f'Connection from {self.client_peer}')
        client_reader, client_writer = await asyncio.open_connection(
            REMOTE_SERVER,
            SERVER_PORT
        )
        print(f'Established connection to {REMOTE_SERVER}:{SERVER_PORT}')
        self.client_writer = client_writer
        self.server_writer = server_writer

        loop = asyncio.get_event_loop()
        loop.create_task(self._receive_client(reader=client_reader))
        loop.create_task(self._receive_server(reader=server_reader))

    async def start_serving(self):
        await asyncio.start_server(self._handle_client, SERVER_BIND, SERVER_PORT)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    server = AsyncioProxy()
    loop.run_until_complete(server.start_serving())
    loop.run_forever()