接下来我们用 python 写下 socks5 服务端的实现
0x03 socks5 实现
先看下整体代码
import logging import socket import struct import select import threading def send_data(sock, data): print(data) bytes_sent = 0 while True: r = sock.send(data[bytes_sent:]) if r < 0: return r bytes_sent += r if bytes_sent == len(data): return bytes_sent def handle_tcp(sock, remote): # 处理 client socket 和 remote socket 的数据流 try: fdset = [sock, remote] while True: # 用 IO 多路复用 select 监听套接字是否有数据流 r, w, e = select.select(fdset, [], []) if sock in r: data = sock.recv(4096) if len(data) <= 0: break result = send_data(remote, data) if result < len(data): raise Exception(failed to send all data) if remote in r: data = remote.recv(4096) if len(data) <= 0: break result = send_data(sock, data) if result < len(data): raise Exception(failed to send all data) except Exception as e: raise(e) finally: sock.close() remote.close() def handle_con(sock, addr): # 接受客户端来的请求，socks5 的 认证和连接过程 sock.recv(256) # 无需进一步认证信息 sock.send(b"\x05\x00") data = sock.recv(4) or \x00 * 4 # CMD 为 0x01 也就是 CONNECT 继续 mode = data[1] if mode != 1: return # DST.ADDR 有三种形式，分别做判断 addr_type = data[3] if addr_type == 1: addr_ip = sock.recv(4) remote_addr = socket.inet_ntoa(addr_ip) elif addr_type == 3: addr_len = int.from_bytes(sock.recv(1), byteorder=big) remote_addr = sock.recv(addr_len) elif addr_type == 4: addr_ip = sock.recv(16) remote_addr = socket.inet_ntop(socket.AF_INET6, addr_ip) else: return # DST.PORT remote_addr_port = struct.unpack(>H, sock.recv(2)) # 返回给客户端 success reply = b"\x05\x00\x00\x01" reply += socket.inet_aton(0.0.0.0) + struct.pack(">H", 8888) sock.send(reply) # 拿到 remote address 的信息后，建立连接 try: remote = socket.create_connection((remote_addr, remote_addr_port[0])) logging.info - 这个网站可出售。 - 最佳的Logging 来源和相关信息。(connecting %s:%d % (remote_addr, remote_addr_port[0])) except socket.error as e: logging.error(e) return handle_tcp(sock, remote) def main(): socketServer = socket.socket(socket.AF_INET, socket.SOCK_STREAM) socketServer.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) socketServer.bind((, 8888)) socketServer.listen(5) try: while True: sock, addr = socketServer.accept() t = threading.Thread(target=handle_con, args=(sock, addr)) t.start() except socket.error as e: logging.error(e) except KeyboardInterrupt: socketServer.close() if __name__ == __main__: main()首先，上面只是个 socks5 服务器的简单实现，处理 TCP 的转发，无需密码认证。
接下来我们分开来看，主要是三个部分 main(), handle_con(), handle_tcp()
main() 函数大家其实挺熟悉的，就是前面教程介绍的 socket 编程的线程方法，绑定套接字监听，然后有客户端请求后，调用 `handle_con()`handle_con 实现了 socks5 的 认证和连接过程，大家对照下注释和前面的 socks5 协议，应该挺容易看懂。认证完成后，就开始转发请求了， handle_tcp() 实现了这个功能，这里面用到了 select ，IO 多路复用模式。其实一开始怕大家不理解，也想用线程来实现，不过发现 IO 多路复用在一个线程中就能监听多个套接字，代码比多线程方式更加简洁，索性就这么用了。这里面没有根据平台去判断用 select, poll 还是 epoll，感兴趣的话可以直接看 ss 的实现