How to create a simple and secure Socks5 proxy in Python?

Socks5 proxy is a network protocol that allows clients to forward network connection requests through a proxy server. Compared with Socks4, Socks5 provides a wider range of authentication methods and address type support, including IPv6 and domain name resolution. Creating a simple and secure Socks5 proxy in Python requires support for authentication and correct protocol handling. Here is a step-by-step guide and code examples:

Step 1: Install necessary libraries

Use the standard library socketserver and struct, no additional installation is required.

Step 2: Write Socks5 proxy server code

import socket
import struct
import select
from socketserver import ThreadingTCPServer, BaseRequestHandler

class Socks5ProxyHandler(BaseRequestHandler):
    username = 'admin'  # Change to a safe username
    password = 'password'  # Change to a strong password

    def handle_auth(self):
        data = self.request.recv(1024)
        if not data or data[0] != 0x05:
            self.request.close()
            return False
        
        # Check whether username and password authentication is supported
        nmethods = data[1]
        methods = data[2:2 + nmethods]
        if 0x02 not in methods:
            self.request.sendall(struct.pack('!BB', 0x05, 0xFF))
            return False
        
        # Select Username/Password Authentication
        self.request.sendall(struct.pack('!BB', 0x05, 0x02))
        
        # Handling Authentication
        auth_data = self.request.recv(1024)
        if not auth_data or auth_data[0] != 0x01:
            return False
        
        ulen = auth_data[1]
        uname = auth_data[2:2 + ulen].decode('utf-8')
        plen = auth_data[2 + ulen]
        passwd = auth_data[3 + ulen:3 + ulen + plen].decode('utf-8')
        
        if uname == self.username and passwd == self.password:
            self.request.sendall(struct.pack('!BB', 0x01, 0x00))
            return True
        else:
            self.request.sendall(struct.pack('!BB', 0x01, 0x01))
            return False

    def handle_request(self):
        # Receiving client requests
        data = self.request.recv(1024)
        if not data or len(data) < 4:
            return False
        
        ver, cmd, _, atyp = struct.unpack('!4B', data[:4])
        if ver != 0x05 or cmd != 0x01:  # Only handle CONNECT requests
            self.request.sendall(struct.pack('!8B', 0x05, 0x07, 0x00, 0x01, 0, 0, 0, 0))
            return False
        
        # Resolve the target address and port
        if atyp == 0x01:  # IPv4
            target_addr = socket.inet_ntop(socket.AF_INET, data[4:8])
            port = struct.unpack('!H', data[8:10])[0]
        elif atyp == 0x03:  # domain name
            domain_len = data[4]
            target_addr = data[5:5 + domain_len].decode('utf-8')
            port = struct.unpack('!H', data[5 + domain_len:5 + domain_len + 2])[0]
        else:
            self.request.sendall(struct.pack('!8B', 0x05, 0x08, 0x00, 0x01, 0, 0, 0, 0))
            return False
        
        # Connect to the target server
        remote = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            remote.connect((target_addr, port))
        except Exception as e:
            self.request.sendall(struct.pack('!8B', 0x05, 0x03, 0x00, 0x01, 0, 0, 0, 0))
            return False
        
        # Response to client connection success
        bind_addr = remote.getsockname()
        response = struct.pack('!4B', 0x05, 0x00, 0x00, 0x01)
        response += socket.inet_aton(bind_addr[0]) + struct.pack('!H', bind_addr[1])
        self.request.sendall(response)
        
        # Data forwarding
        try:
            while True:
                r, _, _ = select.select([self.request, remote], [], [])
                if self.request in r:
                    data = self.request.recv(4096)
                    if not data:
                        break
                    remote.sendall(data)
                if remote in r:
                    data = remote.recv(4096)
                    if not data:
                        break
                    self.request.sendall(data)
        except:
            pass
        finally:
            remote.close()
            return True

    def handle(self):
        if not self.handle_auth():
            return
        self.handle_request()

if __name__ == '__main__':
    # Start the proxy server on local port 1080
    with ThreadingTCPServer(('0.0.0.0', 1080), Socks5ProxyHandler) as server:
        print("Socks5 proxy server started, listening on port 1080...")
        server.serve_forever()

Step 3: Security Enhancements

• Strong Password Policy: Modify the default username and password in the example to a complex combination.
• Restrict Access: Bind to a specific IP (such as 127.0.0.1) instead of 0.0.0.0 to avoid exposure to the public network.
• Log Monitoring: Add logging capabilities to track connection attempts.
• Use TLS Tunnel: Wrap proxy traffic with tools such as stunnel to achieve encrypted transmission.

Step 4: Test the Proxy

Use curl to test whether the proxy is working:

curl --socks5 admin:[email protected]:1080 https://example.com

Notes

• Protocol support: The example only handles TCP CONNECT requests and is applicable to HTTP/HTTPS.
• Performance optimization: For high-concurrency scenarios, it is recommended to use asynchronous libraries such as asyncio.
• Production environment: It is recommended to use mature libraries (such as python-socks) or dedicated proxy software.

Conclusion

This article introduces the need to support authentication and correctly handle protocols to create a simple and secure Socks5 proxy in Python. However, to create a simple and secure Socks5 proxy, you also need to consider the security of authentication mechanisms, encrypted communications, logging, and monitoring.