Forward SOCKSv5 requests to HTTP proxy.

sockshttp.py

#!/usr/bin/env python3
"""Forward SOCKSv5 requests to HTTP proxy.

Frontend: accept SOCKSv5 CONNECT request.
Backend: HTTP proxy with CONNECT support.
"""
import asyncio
import logging
from struct import pack, unpack
from ipaddress import IPv4Address, IPv6Address


FRONTEND = ('127.0.0.1', 10080)
BACKEND = ('192.168.6.65', 3128)
TIMEOUT = 45 * 60

async def handshake_socks(reader, writer):
    """Parse SOCKSv5 request and return (host, port) tuple."""
    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:  # IPv4
        addr = IPv4Address(await reader.readexactly(4))
    elif atype == 0x04:  # IPv6
        addr = IPv6Address(await reader.readexactly(16))
    elif atype == 0x03:  # hostname
        length = await reader.readexactly(1)
        addr = await reader.readexactly(ord(length))
        addr = addr.decode()
    else:
        raise Exception(f'unknown atype {atype}')
    port, = unpack('!H', await reader.readexactly(2))
    writer.write(b'\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00')
    return addr, port


async def open_http_tunnel(addr, port):
    """Open a HTTP CONNECT tunnel with backend server,
    and return (reader, writer) of the tunnel."""
    if isinstance(addr, IPv6Address):
        host = f'[{addr}]:{port}'
    else:
        host = f'{addr}:{port}'
    logging.debug(f'proxy to {host}')

    reader, writer = await asyncio.open_connection(*BACKEND)
    request = f'CONNECT {host} HTTP/1.1\r\nHost: {host}\r\n\r\n'
    writer.write(request.encode())
    line = await reader.readline()
    try:
        version, status, reason = line.decode().strip().split(' ', 2)
    except (ValueError, UnicodeDecodeError):
        raise IOError('unexpected response from backend')
    if version != 'HTTP/1.1':
        raise IOError('unexpected http version: %s', version)
    if not status.startswith('2'):
        raise IOError('backend return error: %s %s', status, reason)
    while (await reader.readline()) != b'\r\n':
        pass
    return reader, writer


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()


async def income_socks(f_reader, f_writer):
    addr, port = await handshake_socks(f_reader, f_writer)
    try:
        b_reader, b_writer = await open_http_tunnel(addr, port)
    except IOError as e:
        logging.warning('fail to connect to backend', exc_info=e)
        f_writer.close()
        return
    logging.info(f'=> {addr},{port}')
    pipings = [piping(f_reader, b_writer), piping(b_reader, f_writer)]
    await asyncio.wait(pipings, return_when=asyncio.FIRST_EXCEPTION)
    f_writer.close()
    b_writer.close()


def main():
    logging.basicConfig(level=logging.DEBUG)
    loop = asyncio.get_event_loop()
    coro = asyncio.start_server(income_socks, *FRONTEND, loop=loop)
    server = loop.run_until_complete(coro)

    logging.info('Serving on {}'.format(server.sockets[0].getsockname()))
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass

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

if __name__ == '__main__':
    main()