Use python asyncio module to unwrap socks5 proxy socket

test_socks5_to_tcp.py

import asyncio
import logging
from asyncio import StreamReader, StreamWriter, StreamReaderProtocol
from typing import Optional

import socks

logging.basicConfig(level=logging.INFO)


async def dial(client_conn, server_conn):
    async def io_copy(reader: StreamReader, writer: StreamWriter):
        while True:
            data = await reader.read(8192)
            if not data:
                break
            writer.write(data)
        writer.close()

    asyncio.ensure_future(io_copy(client_conn[0], server_conn[1]))
    asyncio.ensure_future(io_copy(server_conn[0], client_conn[1]))


async def open_socks5_connection(host: str, port: int,
                                 *,
                                 username: Optional[str] = None, password: Optional[str] = None,
                                 socks_host: str = 'localhost', socks_port: int = 1080,
                                 limit=2 ** 16, loop=None):
    # create socks5 socket and connect to target server
    s = socks.socksocket()
    s.set_proxy(socks.SOCKS5, addr=socks_host, port=socks_port, username=username, password=password)
    s.connect((host, port))

    # convert sockssocket to asyncio Stream{Reader,Writer}
    if not loop:
        loop = asyncio.get_event_loop()

    reader = StreamReader(limit=limit, loop=loop)
    protocol = StreamReaderProtocol(reader, loop=loop)
    transport, _ = await loop.create_connection(lambda: protocol, sock=s)
    writer = StreamWriter(transport, protocol, reader, loop)
    return reader, writer


async def handle_simple_obfs_connection(reader, writer):
    server_conn = await open_socks5_connection(host='www.google.com', port=80, socks_host='127.0.0.1', socks_port=1086)
    asyncio.ensure_future(dial((reader, writer), server_conn))


def main():
    loop = asyncio.get_event_loop()
    server = asyncio.start_server(handle_simple_obfs_connection, host='127.0.0.1', port=40001)
    try:
        server = loop.run_until_complete(server)
        loop.run_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.close()


if __name__ == '__main__':
    main()