mitmproxy_wireguard_server_tcp_udp_echo_handler.py

1import asyncio
2import mitmproxy_wireguard
3
4async def main():
5    # A basic handler that just logs and echoes back data.
6    class EchoHandler:
7        def handle_tcp(self, reader, writer):
8            async def pipe():
9                while data := await reader.read(1024):
10                    writer.write(data)
11                    await writer.drain()
12                writer.close()
13                await writer.wait_closed()
14            asyncio.create_task(pipe())
15
16        def handle_udp(self, transport, data, remote_addr):
17            transport.sendto(data, remote_addr)
18
19    # Start the WireGuard server.
20    # Replace the keys and addresses with your own configuration.
21    server = await mitmproxy_wireguard.start_server(
22        host="0.0.0.0",
23        port=51820,
24        private_key="YOUR_SERVER_PRIVATE_KEY",
25        peer_public_keys=["PEER_PUBLIC_KEY"],
26        handle_connection=EchoHandler().handle_tcp,
27        handle_datagram=EchoHandler().handle_udp,
28    )
29
30    print("WireGuard server is running...")
31    try:
32        await asyncio.Future()  # run forever
33    finally:
34        server.close()
35        await server.wait_closed()
36
37if __name__ == "__main__":
38    asyncio.run(main())