Skip to content

Instantly share code, notes, and snippets.

@alfonsrv
Created June 5, 2022 15:01
Show Gist options
  • Save alfonsrv/97d23b4d2e7ceb328f5fc6e7fb7d4478 to your computer and use it in GitHub Desktop.
Save alfonsrv/97d23b4d2e7ceb328f5fc6e7fb7d4478 to your computer and use it in GitHub Desktop.
asyncio Man-in-the-Middle Proxy – allowing for bidirectional streaming
"""
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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment