multi I/O traffic forward. using select/epoll or asyncio lib

proxy_asyncio.py

import asyncio


async def proxy_pass(reader, writer):
    socks_r, socks_w = await asyncio.open_connection('127.0.0.1', 1080) 
    
    async def a_2_b(reader, writer):
        data = await reader.read(4096)
        while 1:
            try:                
                if data:
                    writer.write(data)
                    await writer.drain()
                    data = await reader.read(4096)
                else:
                    writer.close()
                    break
            except ConnectionResetError:
                writer.close()
                break


    asyncio.run_coroutine_threadsafe(a_2_b(reader, socks_w), loop)
    asyncio.run_coroutine_threadsafe(a_2_b(socks_r, writer), loop)


loop = asyncio.get_event_loop()
coro = asyncio.start_server(proxy_pass, '0.0.0.0', 2333)
server = loop.run_until_complete(coro)
print('[+]Server listen on 0.0.0.0:2333')
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

proxy_epoll.py

import socket
import selectors
import random
import struct


PROXY_POOL = [('127.0.0.1', 1080)]

def proxy_pass(server):
    client_conn, addr = server.accept()
    print(f'[+]get client {addr[0]}:{addr[1]}')
    client_conn.setblocking(False)
    # set socket SO_LINGER, to make sure closing socket
    client_conn.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, 
                          struct.pack('ii', 1, 0))
    
    socks5_conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    socks5_conn.setblocking(False)
    socks5_conn.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
                          struct.pack('ii', 1, 0))
    try:
        socks5_conn.connect(random.choice(PROXY_POOL))
    except BlockingIOError:
        pass
    done = False
    
    def on_socks_connected():
        s.unregister(socks5_conn)
        s.register(client_conn, selectors.EVENT_READ, client_2_socks5)
        s.register(socks5_conn, selectors.EVENT_READ, socks5_2_client)
        
    def client_2_socks5():
        nonlocal done
        if done:
            print(f'[+]handle {addr[0]}:{addr[1]} over')
            s.unregister(client_conn)
            client_conn.close()
            return        
        try:
            data = client_conn.recv(4096)
            if data:
                socks5_conn.sendall(data)
                print(f'client ====> proxy ====> socks5 :: bytes:{len(data)}')                
            else:
                done = True
        except (BlockingIOError, ConnectionResetError):
            done = True
    
    def socks5_2_client():
        nonlocal done        
        if done:
            print(f'[+]handle {addr[0]}:{addr[1]} over')            
            s.unregister(socks5_conn)
            socks5_conn.close()
            return        
        try:
            data = socks5_conn.recv(4096)
            if data:
                client_conn.sendall(data)
                print(f'socks5 ====> proxy ====> client :: bytes:{len(data)}')
            else:
                done = True
        except (BlockingIOError, ConnectionResetError):
            done = True
            
    s.register(socks5_conn, selectors.EVENT_WRITE, on_socks_connected)
       

s = selectors.DefaultSelector()
proxy_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
proxy_server.bind(('0.0.0.0', 2333))

proxy_server.listen(32)
proxy_server.setblocking(False)
s.register(proxy_server, selectors.EVENT_READ, proxy_pass)

print('[+]Server listen on 0.0.0.0:2333')
while 1:
    events = s.select()
    for key, _ in events:
        callback = key.data
        if callback is proxy_pass:
            callback(key.fileobj)
        else:
            callback()