基于Python的跨NAT反向代理故障修复请求

问题背景 #

家中有一台处于NAT后的个人PC（动态IP），搭配一台拥有公网IP的VPS，需要实现通过VPS中转，从任意地点使用家网IP访问网络。现有Python脚本可初始运行，但打开1-2个网站后，服务器停止监听代理请求，无法转发数据包。

问题根源分析 #

1. 线程安全问题：共享数据结构array被多个线程同时读写，未加锁，导致数据丢失或线程死循环。
2. 忙等浪费资源：使用while True循环轮询数据，占用大量CPU，且可能导致线程阻塞。
3. 数据包解析错误：客户端域名解析时索引计算错误，导致域名读取异常。
4. 连接处理不严谨：连接关闭时未正确终止线程，资源未完全释放；服务器端接收preamble时未处理数据包拆分情况。

修复后的代码 #

服务器端代码 #

import socket
import threading
from threading import Condition

class Proxy:
    def __init__(self):
        self.client = None
        self.uid = 0
        self.array = []
        self.array_lock = Condition()  # 线程安全的条件变量，替代忙等
        self.username = 'username'
        self.password = 'password'
    
    def get_available_methods(self, nmethods, connection):
        methods = []
        for i in range(nmethods):
            method = connection.recv(1)
            if not method:
                return []
            methods.append(ord(method))
        return methods
        
    def verify_credentials(self, connection):
        version_byte = connection.recv(1)
        if not version_byte:
            connection.close()
            return False
        version = ord(version_byte)

        username_len_byte = connection.recv(1)
        if not username_len_byte:
            connection.close()
            return False
        username_len = ord(username_len_byte)
        
        username = connection.recv(username_len)
        if not username:
            connection.close()
            return False
        username = username.decode('utf-8')

        password_len_byte = connection.recv(1)
        if not password_len_byte:
            connection.close()
            return False
        password_len = ord(password_len_byte)
        
        password = connection.recv(password_len)
        if not password:
            connection.close()
            return False
        password = password.decode('utf-8')

        if username == self.username and password == self.password:
            response = bytes([version, 0])
            connection.sendall(response)
            return True

        response = bytes([version, 0xFF])
        connection.sendall(response)
        connection.close()
        return False

    def handle_admin(self, connection, identifier):
        try:
            initial_data = connection.recv(2)
            if len(initial_data) != 2:
                connection.close()
                return
            version, nmethods = initial_data

            methods = self.get_available_methods(nmethods, connection)

            if 2 not in set(methods):
                connection.close()
                return

            connection.sendall(bytes([5, 2]))

            if not self.verify_credentials(connection):
                return
                
            preamble = b'\xFF\xFE\x00\xFF'
            self.client.sendall(preamble + identifier.to_bytes(4, 'big'))

            cmd_data = connection.recv(4)
            if len(cmd_data) != 4:
                connection.close()
                return
            version, cmd, _, address_type = cmd_data
      
            address = None
            if address_type == 1:  # IPv4
                addr_bytes = connection.recv(4)
                if len(addr_bytes) !=4:
                    connection.close()
                    return
                address = socket.inet_ntoa(addr_bytes)
            elif address_type == 3:  # Domain name
                domain_len_byte = connection.recv(1)
                if not domain_len_byte:
                    connection.close()
                    return
                domain_length = ord(domain_len_byte)
                domain = connection.recv(domain_length)
                if not domain:
                    connection.close()
                    return
                address = socket.gethostbyname(domain.decode('utf-8'))
            else:
                connection.close()
                return
                
            port_data = connection.recv(2)
            if len(port_data) !=2:
                connection.close()
                return
            port = int.from_bytes(port_data, 'big', signed=False)
            
            print(f'[Client] {address}:{port}')
            
            preamble = b'\xFF\xFE\xFF\x00'
            if address_type == 1:  # IPv4
                addr_bytes = socket.inet_aton(address)
            elif address_type == 3:  # Domain name
                addr_bytes = domain_len_byte + domain
            port_bytes = port.to_bytes(2, 'big')
            
            packet2 = port_bytes + cmd.to_bytes(2, 'big') + address_type.to_bytes(1, 'big') + addr_bytes
            self.client.sendall(preamble + identifier.to_bytes(4, 'big') + len(packet2).to_bytes(4, 'big') + packet2)
            
            # 等待客户端返回的响应数据
            found_data = None
            with self.array_lock:
                while True:
                    for idx, (number, data) in enumerate(self.array):
                        if number == identifier:
                            found_data = data
                            del self.array[idx]
                            break
                    if found_data:
                        break
                    self.array_lock.wait()  # 释放锁等待，避免忙等
            
            connection.sendall(found_data)
            print(f'Reply sent, len: {len(found_data)}')

            t = threading.Thread(target=self.get_connection, args=(connection, identifier), daemon=True)
            t.start()
            
            go_break = False
            while True:
                to_remove = []
                with self.array_lock:
                    for idx, (number, data) in enumerate(self.array):
                        if number == identifier:
                            to_remove.append((idx, data))  
                    
                    # 先收集再删除，避免遍历中修改列表
                    for idx, data in to_remove:
                        del self.array[idx]
                
                if not to_remove:
                    with self.array_lock:
                        # 等待新数据，超时检查连接状态
                        if not self.array_lock.wait(0.5):
                            # 检查连接是否存活
                            try:
                                connection.send(b'')  # 空包测试连接
                            except:
                                go_break = True
                                break
                    continue
                
                print(f'Get from server && Reply to admin {len(data)} bytes.')
                try:
                    for _, data in to_remove:
                        connection.sendall(data)
                except Exception as text:
                    print(f'[connection.sendall(data)] Exception! {text}')
                    go_break = True
                    break
        
        except Exception as text:
            print(f'[handle_admin] Exception! {text}')
        finally:
            try:
                connection.close()
            except Exception as text:
                print(f'[connection.close()] Exception! {text}')
        
    def get_connection(self, connection, identifier):
        try:
            while True:
                data = connection.recv(40960)
                if not data:
                    break
                print(f'Get from admin && Send to client {len(data)} bytes')
                self.client.sendall(b'\xFF\xFE\xFF\x00' + identifier.to_bytes(4, 'big') + len(data).to_bytes(4, 'big') + data)
        except Exception as text:
            print(f'[get_connection] Exception! {text}')
        finally:
            try:
                connection.close()
            except:
                pass
        
    def connect_to_client(self, host, port):
        client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client.bind((host, port))
        client.listen(5)
        
        print(f'[Client] SOCKS5 proxy server is running on {host}:{port}')
        
        connection, address = client.accept()
        print(f'[Client] New connection from {address}')
        try:
            while True:
                # 确保接收到完整的2字节preamble
                preamble = b''
                while len(preamble) <2:
                    chunk = connection.recv(2 - len(preamble))
                    if not chunk:
                        raise ConnectionResetError("Client disconnected")
                    preamble += chunk
                
                if preamble == b'\xFF\xFE':
                    if not self.client:
                        self.client = connection
                        print('The client has been successfully authorized!')
                    else:
                        # 接收第二个preamble部分
                        preamble2 = b''
                        while len(preamble2) <2:
                            chunk = connection.recv(2 - len(preamble2))
                            if not chunk:
                                raise ConnectionResetError("Client disconnected")
                            preamble2 += chunk
                        
                        if preamble2 == b'\x00\xFF':
                            # 接收number
                            number_bytes = b''
                            while len(number_bytes) <4:
                                chunk = connection.recv(4 - len(number_bytes))
                                if not chunk:
                                    raise ConnectionResetError("Client disconnected")
                                number_bytes += chunk
                            number = int.from_bytes(number_bytes, 'big')
                            
                            # 接收size
                            size_bytes = b''
                            while len(size_bytes) <4:
                                chunk = connection.recv(4 - len(size_bytes))
                                if not chunk:
                                    raise ConnectionResetError("Client disconnected")
                                size_bytes += chunk
                            size = int.from_bytes(size_bytes, 'big')
                            
                            print(f'New data! Size: {size}')
                            data = b''
                            while len(data) < size:
                                packet = connection.recv(size - len(data))
                                if not packet:
                                    raise ConnectionResetError("Client disconnected")
                                data += packet
                            print(f'Len: {len(data)}')
                            
                            with self.array_lock:
                                self.array.append((number, data))
                                self.array_lock.notify_all()  # 通知等待的线程
        except Exception as e:
            print(f'[connect_to_client] Error: {e}')
        finally:
            try:
                connection.close()
                self.client = None
            except:
                pass
        
    def run(self, host, port):
        admin = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        admin.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        admin.bind((host, port))
        admin.listen(10)

        print(f'[Admin] SOCKS5 proxy server is running on {host}:{port}')

        while True:
            try:
                connection, address = admin.accept()
                print(f'[Admin] New connection from {address}')
                if self.client:
                    self.uid += 1
                    t = threading.Thread(target=self.handle_admin, args=(connection, self.uid), daemon=True)
                    t.start()
                else:
                    connection.close()
                    print('[Admin] No client connected yet, rejected')
            except Exception as e:
                print(f'[run] Error: {e}')

if __name__ == '__main__':
    proxy = Proxy()
    t = threading.Thread(target=proxy.connect_to_client, args=('0.0.0.0', 55555), daemon=True)
    t.start()
    proxy.run('0.0.0.0', 44444)

客户端代码 #

import socket
import threading
from threading import Condition

class Proxy:
    def __init__(self):
        self.array = []
        self.array_lock = Condition()

    def request(self, server, identifier):
        try:
            # 等待服务器发来的请求数据
            found_data = None
            with self.array_lock:
                while True:
                    for idx, (number, data) in enumerate(self.array):
                        if number == identifier:
                            found_data = data
                            del self.array[idx]
                            break
                    if found_data:
                        break
                    self.array_lock.wait()
            
            port = int.from_bytes(found_data[0:2], 'big')
            cmd = int.from_bytes(found_data[2:4], 'big')
            address_type = int.from_bytes(found_data[4:5], 'big')
            address = None
            
            if address_type == 1:
                addr_bytes = found_data[5:9]
                address = socket.inet_ntoa(addr_bytes)
            elif address_type == 3:
                domain_length = int.from_bytes(found_data[5:6], 'big')
                # 修复索引错误：原代码是6:domain_length，应该是6:6+domain_length
                domain = found_data[6:6+domain_length].decode('utf-8')
                address = socket.gethostbyname(domain)
            else:
                print(f'Unsupported address type: {address_type}')
                return
            
            print(f'Successfully get: {address} {port} | {cmd}')
            
            remote = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            remote.connect((address, port))
            bind_address = remote.getsockname()
            
            reply_address = socket.inet_aton(bind_address[0])
            reply_port = bind_address[1]

            reply = b''.join([
                b'\x05',  # SOCKS5版本
                b'\x00',  # 成功响应
                b'\x00',  # RSV
                b'\x01',  # IPv4地址类型
                reply_address,
                reply_port.to_bytes(2, 'big')
            ])
            preamble = b'\xFF\xFE\x00\xFF'
            server.sendall(preamble + identifier.to_bytes(4, 'big') + len(reply).to_bytes(4, 'big') + reply)
            print(f'Send reply, len: #{len(reply)}')
            
            if reply[1] == 0 and cmd == 1:
                t = threading.Thread(target=self.remote_get, args=(remote, server, identifier), daemon=True)
                t.start()
                
                go_break = False
                while True:
                    to_remove = []
                    with self.array_lock:
                        for idx, (number, data) in enumerate(self.array):
                            if number == identifier:
                                to_remove.append((idx, data))
                        
                        for idx, data in to_remove:
                            del self.array[idx]
                    
                    if not to_remove:
                        with self.array_lock:
                            if not self.array_lock.wait(0.5):
                                # 检查远程连接是否存活
                                try:
                                    remote.send(b'')
                                except:
                                    go_break = True
                                    break
                        continue
                    
                    try:
                        for _, data in to_remove:
                            print(f'Get from server && Reply to remote {len(data)} bytes')
                            remote.sendall(data)
                    except Exception as text:
                        print(f'[remote.sendall(data)] [{identifier}] Exception! {text}')
                        go_break = True
                        break
        except Exception as e:
            print(f'[request] Error: {e}')
        finally:
            try:
                remote.close()
            except:
                pass
          
    def remote_get(self, remote, server, identifier):
        try:
            while True:
                data = remote.recv(40960)
                if not data:
                    break
                print(f'Get from remote && Send to server {len(data)} bytes')
                server.sendall(b'\xFF\xFE\x00\xFF' + identifier.to_bytes(4, 'big') + len(data).to_bytes(4, 'big') + data)
        except Exception as e:
            print(f'[remote_get] Error: {e}')
        finally:
            try:
                remote.close()
            except:
                pass
          
    def listen_for_messages(self, server):
        try:
            while True:
                # 接收preamble
                preamble = b''
                while len(preamble) <2:
                    chunk = server.recv(2 - len(preamble))
                    if not chunk:
                        print('Connection closed.')
                        return
                    preamble += chunk
                
                if preamble == b'\xFF\xFE':
                    preamble2 = b''
                    while len(preamble2) <2:
                        chunk = server.recv(2 - len(preamble2))
                        if not chunk:
                            print('Connection closed.')
                            return
                        preamble2 += chunk
                    
                    if preamble2 == b'\x00\xFF':
                        number_bytes = b''
                        while len(number_bytes) <4:
                            chunk = server.recv(4 - len(number_bytes))
                            if not chunk:
                                print('Connection closed.')
                                return
                            number_bytes += chunk
                        number = int.from_bytes(number_bytes, 'big')
                        t = threading.Thread(target=self.request, args=(server, number), daemon=True)
                        t.start()
                        print('New thread!')
                    elif preamble2 == b'\xFF\x00':
                        number_bytes = b''
                        while len(number_bytes) <4:
                            chunk = server.recv(4 - len(number_bytes))
                            if not chunk:
                                print('Connection closed.')
                                return
                            number_bytes += chunk
                        number = int.from_bytes(number_bytes, 'big')
                        
                        size_bytes = b''
                        while len(size_bytes) <4:
                            chunk = server.recv(4 - len(size_bytes))
                            if not chunk:
                                print('Connection closed.')
                                return
                            size_bytes += chunk
                        size = int.from