A simple SOCKSv5 wrapper for obfs4proxy.

obfs4-client.py

#!/usr/bin/env python3
import asyncio
from struct import pack
from ipaddress import ip_address


TIMEOUT = 300
OBFS_BIN = '/usr/bin/obfs4proxy'
OBFS_EXEC_PARAMS = ['-enableLogging', '-logLevel', 'DEBUG']

# Paste obfs4 server's cert below.
# Params separated by semicolon (;), not comma.
OBFS_CONN_PARAMS = 'cert=xxxxxxxxxxxxxxxxxx;iat-mode=1'

# obfs4 server address & port. IPv4 address only, domain name not supported.
REMOTE_HOST = '1.2.3.4'
REMOTE_PORT = 1984

# Local SOCKSv5 listen on:
BIND_ADDR = '127.0.0.1'
BIND_PORT = 8800

def main():
    loop = asyncio.get_event_loop()
    obfs_addr = loop.run_until_complete(start_obfs())
    print(f'obfs listen on {obfs_addr}')

    coro = asyncio.start_server(
        partial(handle_conn, obfs_addr),
        BIND_ADDR,
        BIND_PORT,
        loop=loop
    )
    server = loop.run_until_complete(coro)
    print(f'socks5 listen on {BIND_ADDR}:{BIND_PORT}')
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass

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


async def start_obfs():
    env = {
        'TOR_PT_MANAGED_TRANSPORT_VER': '1',
        'TOR_PT_EXIT_ON_STDIN_CLOSE': '1',
        'TOR_PT_CLIENT_TRANSPORTS': 'obfs4',
        'TOR_PT_STATE_LOCATION': './client',
    }
    obfs = await asyncio.create_subprocess_exec(
        OBFS_BIN,
        *OBFS_EXEC_PARAMS,
        env=env,
        stdout=asyncio.subprocess.PIPE
    )
    while True:
        line = await obfs.stdout.readline()
        print(line)
        cols = line.decode().split()
        if cols[0] == 'CMETHOD':
            addr = cols[-1].split(':')
            return addr[0], int(addr[1])
        elif cols[0] == 'CMETHOD-ERROR':
            print(line)
            return

async def handle_conn(obfs_addr, reader, writer):
    remote_reader, remote_writer = await handshake_pt(*obfs_addr)
    print('pt conn created')
    request = await handshake_socks(reader, writer)
    remote_writer.write(request)
    pipings = [piping(reader, remote_writer), piping(remote_reader, writer)]
    await asyncio.wait(pipings, return_when=asyncio.FIRST_EXCEPTION)
    writer.close()
    remote_writer.close()


async def handshake_pt(addr, port):
    reader, writer = await asyncio.open_connection(addr, port)
    writer.write(b'\x05\x01\x02')
    method = await reader.readexactly(2)
    assert method == b'\x05\x02'
    writer.write(pack('!BB', 1, len(OBFS_CONN_PARAMS)) + OBFS_CONN_PARAMS.encode() + b'\x01\x00')
    status = await reader.readexactly(2)
    assert status == b'\x01\x00'

    addr = ip_address(REMOTE_HOST)
    writer.write(b'\x05\x01\x00\x01' + addr.packed + pack('!H', REMOTE_PORT))
    resp = await reader.readexactly(10)
    assert resp[:2] == b'\x05\x00'
    return reader, writer


async def handshake_socks(reader, writer):
    auth = await reader.readexactly(2)
    assert auth[0] == 0x05
    await reader.readexactly(auth[1])
    writer.write(b'\x05\x00')
    request = await reader.readexactly(4)
    assert request.startswith(b'\x05\x01')
    atype = request[-1]
    if atype == 0x01:
        request = bytes([atype]) + await reader.readexactly(6)
    elif atype == 0x04:
        request = bytes([atype]) + await reader.readexactly(18)
    elif atype == 0x03:
        length = await reader.readexactly(1)
        request = bytes([atype]) + length + \
                await reader.readexactly(ord(length) + 2)
    else:
        raise Exception(f'unknown atype {atype}')
    writer.write(b'\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00')
    return request


async def piping(reader, writer):
    while True:
        data = await asyncio.wait_for(reader.read(8196), TIMEOUT)
        if not data:
            if writer.can_write_eof():
                writer.write_eof()
            return
        writer.write(data)
        await writer.drain()


if __name__ == '__main__':
    main()

obfs4-server.py

#!/usr/bin/env python3
import asyncio
from struct import unpack
from ipaddress import IPv4Address, IPv6Address


TIMEOUT = 300
OBFS_BIN = '/usr/bin/obfs4proxy'
OBFS_EXEC_PARAMS = ['-enableLogging', '-logLevel', 'DEBUG']

BIND_ADDR = '0.0.0.0'
BIND_PORT = 1984

def main():
    loop = asyncio.get_event_loop()
    coro = asyncio.start_server(
        handle_conn,
        '127.0.0.1',
        loop=loop
    )
    server = loop.run_until_complete(coro)
    proxy_addr = server.sockets[0].getsockname()
    print(f'internal proxy listen on {proxy_addr}')

    loop.run_until_complete(start_obfs(*proxy_addr))
    print(f'obfs4 listen on {BIND_ADDR}:{BIND_PORT}')
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass

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


async def start_obfs(addr, port):
    env = {
        'TOR_PT_MANAGED_TRANSPORT_VER': '1',
        'TOR_PT_EXIT_ON_STDIN_CLOSE': '1',
        'TOR_PT_STATE_LOCATION': './server',
        'TOR_PT_SERVER_TRANSPORTS': 'obfs4',
        'TOR_PT_SERVER_BINDADDR': f'obfs4-{BIND_ADDR}:{BIND_PORT}',
        'TOR_PT_ORPORT': f'{addr}:{port}',
    }
    obfs = await asyncio.create_subprocess_exec(
        OBFS_BIN,
        *OBFS_EXEC_PARAMS,
        env=env,
        stdout=asyncio.subprocess.PIPE
    )
    while True:
        line = (await obfs.stdout.readline()).decode().strip()
        print(line)
        if line.startswith('SMETHOD-ERROR') or not line:
            raise Exception('cannot start obfs')
        if line.startswith('SMETHODS DONE'):
            return

async def handle_conn(reader, writer):
    addr, port = await parse_request(reader)
    print(f'connect to {addr}:{port}')
    remote_reader, remote_writer = await asyncio.open_connection(addr, port)
    pipings = [piping(reader, remote_writer), piping(remote_reader, writer)]
    await asyncio.wait(pipings, return_when=asyncio.FIRST_EXCEPTION)
    writer.close()
    remote_writer.close()


async def parse_request(reader):
    atype = ord(await reader.readexactly(1))
    if atype == 0x01:
        addr = IPv4Address(await reader.readexactly(4))
    elif atype == 0x04:
        addr = IPv6Address(await reader.readexactly(16))
    elif atype == 0x03:
        length = ord(await reader.readexactly(1))
        addr = (await reader.readexactly(length)).decode()
    else:
        raise Exception(f'Unknown atype {atype}')
    port, = unpack('!H', await reader.readexactly(2))
    return addr, port


async def piping(reader, writer):
    while True:
        data = await asyncio.wait_for(reader.read(8196), TIMEOUT)
        if not data:
            if writer.can_write_eof():
                writer.write_eof()
            return
        writer.write(data)
        await writer.drain()


if __name__ == '__main__':
    main()