实现 socks5 代理

# -*- coding: utf-8 -*-

import socket
import threading

source_host = '127.0.0.1'
source_port = 5002

desc_host = '127.0.0.1'
desc_port = 8099


def send(sender, recver):
    while 1:
        try:
            data = sender.recv(2048)
        except:
            print("recv error")
            break
        if not data:
            break

        try:
            recver.sendall(data)
        except:
            print("send error")
            break
    sender.close()
    recver.close()


def proxy(client):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.connect((source_host, source_port))
    threading.Thread(target=send, args=(client, server)).start()
    threading.Thread(target=send, args=(server, client)).start()


def main():
    proxy_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    proxy_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    proxy_server.bind((desc_host, desc_port))
    proxy_server.listen(5)

    print("Proxying from %s:%s to %s:%s ..." % (source_host, source_port, desc_host, desc_port))

    while 1:
        conn, addr = proxy_server.accept()
        print("received connect from %s:%s" % (addr[0], addr[1]))
        threading.Thread(target=proxy, args=(conn,)).start()


if __name__ == '__main__':
    main()

# -*- coding: utf-8 -*-
import socket
import logging
import threading
from socketserver import StreamRequestHandler, BaseRequestHandler
from socketserver import ThreadingTCPServer as RawThreadingTCPServer

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class ThreadingTCPServer(RawThreadingTCPServer):
    allow_reuse_address = True


class ProxyHandler(BaseRequestHandler):
    def handle(self):
        # should be blocked until finish handle.
        target = socket.create_connection(('127.0.0.1', 5002))
        thread_list = []
        thread_list.append(threading.Thread(
            target=self.forward_reply, args=(self.request, target)
        ))
        thread_list.append(threading.Thread(
            target=self.forward_reply, args=(target, self.request)
        ))
        for t in thread_list:
            t.start()
        for t in thread_list:
            t.join()

    def forward_reply(self, local, target):
        try:
            while True:
                # target.write(local.read())
                data = local.recv(1024)
                if not data:
                    return
                target.send(data)
        except Exception as e:
            logger.info("forward error:%r", e)
            raise e


if __name__ == '__main__':
    server = ThreadingTCPServer(('127.0.0.1', 5003), ProxyHandler)
    server.serve_forever()

# -*- coding: utf-8 -*-
import socket
import struct
import select
import gevent
import logging
import threading
from socketserver import StreamRequestHandler
from socketserver import ThreadingTCPServer as RawThreadingTCPServer
from gevent import monkey

monkey.patch_all()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
AUTH_USERNAME = "YOUR_PROXY_LOGIN"
AUTH_PASSWORD = "YOUR_PROXY_PASSWORD"


class ThreadingTCPServer(RawThreadingTCPServer):
    allow_reuse_address = True


class SocksException(Exception):
    pass


class Socks5Handler(StreamRequestHandler):
    def handle_forward(self, client, remote):
        # two threading to read and write
        # handle should be blocked
        thread_list = [threading.Thread(
            target=self.forward_reply, args=(client, remote)
        ), threading.Thread(
            target=self.forward_reply, args=(remote, client)
        )]
        for t in thread_list:
            t.start()
        for t in thread_list:
            t.join()

    def handle_forward_in_gevent(self, client, remote):
        spawns = [gevent.spawn(self.forward_reply, client, remote), gevent.spawn(self.forward_reply, remote, client)]
        gevent.joinall(spawns)

    def forward_reply(self, local, target):
        # type: (socket.socket, socket.socket) -> None
        try:
            while True:
                # target.write(local.read())
                data = local.recv(1024)
                if not data:
                    return
                target.send(data)
        except Exception as e:
            logger.info("forward error:%r", e)
            raise e

    def handle_forward_in_select(self, client, remote):
        # type: (socket.socket, socket.socket) -> None
        try:
            inputs = [client, remote]
            while inputs:
                readable, _, _ = select.select(inputs, [], [])
                if client in readable:
                    data = client.recv(4096)
                    if not data:
                        break
                    remote.sendall(data)

                if remote in readable:
                    data = remote.recv(4096)
                    if not data:
                        break
                    client.sendall(data)
        except Exception as e:
            logger.info("select error:%r", e)

    def socks_user_auth(self):
        buf = self.rfile.read(2)
        if buf[0] != chr(0x01):
            raise SocksException("invalid version")
        username_length = ord(buf[1])
        username = self.rfile.read(username_length)

        buf = self.rfile.read(1)
        password_length = ord(buf[0])
        password = self.rfile.read(password_length)
        logger.info("user auth: %s-%s", username, password)

        if username == AUTH_USERNAME and password == AUTH_PASSWORD:
            self.wfile.write(b'\x01\x00')
        else:
            self.wfile.write(b'\x01\x01')
            raise SocksException("socks user auth fail")

    def socks_auth(self):
        # self.rfile._sock.recv(2)
        buf = self.rfile.read(2)
        if buf[0] != chr(0x05):
            # python3 will get int
            # if buf[0] != chr(0x05):
            # if buf[0] != b'\x05':
            raise SocksException("invalid version")

        methods_length = ord(buf[1])
        methods = self.rfile.read(methods_length)
        logger.info("receive methods:%r", methods)
        if b'\x02' in methods:
            # socks user auth
            self.wfile.write(b'\x05\x02')
            self.socks_user_auth()
        else:
            self.wfile.write(b'\x05\x00')
        # maybe need one
        # self.wfile.flush()

    def socks_connect(self):
        ver, cmd, _, addr_type = self.rfile.read(4)
        if ver != chr(0x05) or cmd != chr(0x01):
            raise SocksException("invalid ver/cmd")

        if addr_type == chr(0x01):
            logger.info("client addr type:IPv4")
            packed_ip = self.rfile.read(4)
            remote_addr = socket.inet_ntoa(packed_ip)
        elif addr_type == chr(0x03):
            logger.info("client addr type:URL")
            addr_length = self.wfile.read(1)
            remote_addr = self.rfile.read(ord(addr_length))
        elif addr_type == chr(0x04):
            logger.info("client addr type:IPv6")
            packed_ip = self.rfile.read(16)
            remote_addr = socket.inet_ntop(socket.AF_INET6, packed_ip)
        else:
            raise SocksException("addr_type not supported.")

        addr_port = self.rfile.read(2)
        # 注意：这里是返回的tuple
        remote_port = struct.unpack('>H', addr_port)[0]

        reply = b'\x05\x00\x00\x01'
        reply += socket.inet_aton('0.0.0.0') + struct.pack('>H', 2222)
        self.wfile.write(reply)
        return remote_addr, remote_port

    def handle(self):
        try:
            logger.info("receive from:%r", self.request.getpeername())
            # step one: sock auth
            self.socks_auth()

            # step two: sock connect
            remote_addr, remote_port = self.socks_connect()
            logger.info("target addr:%r", (remote_addr, remote_port))

            # step three: sock forward
            remote = socket.create_connection((remote_addr, remote_port))
            self.handle_forward_in_select(self.connection, remote)
        except SocksException as e:
            logger.info("socks error:%r", e)
        except socket.error as e:
            logger.info("socket error:%r", e)
            raise


if __name__ == '__main__':
    server = ThreadingTCPServer(('127.0.0.1', 8572), Socks5Handler)
    server.serve_forever()