破壁者指南：内网穿透技术的深度解构与实战方法

# NAT类型检测算法实现
import socket
import struct
import threading
import time

class NATTypeDetector:
    """NAT类型检测器 - 基于STUN协议实现"""
    
    def __init__(self, stun_servers=None):
        self.stun_servers = stun_servers or [
            ('stun.l.google.com', 19302),
            ('stun1.l.google.com', 19302),
            ('stun2.l.google.com', 19302)
        ]
        self.local_ip = self._get_local_ip()
        
    def _get_local_ip(self):
        """获取本地IP地址"""
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.connect(('8.8.8.8', 80))
            local_ip = s.getsockname()[0]
            s.close()
            return local_ip
        except Exception:
            return '127.0.0.1'
    
    def detect_nat_type(self):
        """检测NAT类型"""
        results = []
        
        for server in self.stun_servers:
            try:
                result = self._stun_test(server)
                if result:
                    results.append(result)
            except Exception as e:
                print(f"STUN测试失败 {server}: {e}")
                
        return self._analyze_results(results)
    
    def _stun_test(self, server):
        """执行STUN测试"""
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.settimeout(5)
        
        try:
            # 构造STUN Binding Request
            transaction_id = b'\x00' * 12
            stun_request = b'\x00\x01\x00\x00' + transaction_id
            
            sock.sendto(stun_request, server)
            response, addr = sock.recvfrom(1024)
            
            # 解析STUN响应
            if len(response) >= 20:
                mapped_addr = self._parse_mapped_address(response)
                return {
                    'server': server,
                    'local_addr': (self.local_ip, sock.getsockname()[1]),
                    'mapped_addr': mapped_addr,
                    'server_addr': addr
                }
        finally:
            sock.close()
            
        return None

flowchart TD
    A[内网穿透技术] --> B[直接连接类]
    A --> C[打洞技术类]
    A --> D[中继转发类]
    A --> E[协议辅助类]
    
    B --> B1[UPnP端口映射]
    B --> B2[NAT-PMP协议]
    B --> B3[手动端口转发]
    
    C --> C1[UDP打洞]
    C --> C2[TCP打洞]
    C --> C3[ICMP打洞]
    
    D --> D1[TURN中继]
    D --> D2[HTTP隧道]
    D --> D3[WebSocket隧道]
    
    E --> E1[STUN协议]
    E --> E2[ICE框架]
    E --> E3[SDP协商]
    
    classDef directClass fill:#e1f5fe,stroke:#01579b,stroke-width:2px
    classDef holeClass fill:#f3e5f5,stroke:#4a148c,stroke-width:2px
    classDef relayClass fill:#fff3e0,stroke:#e65100,stroke-width:2px
    classDef protocolClass fill:#e8f5e8,stroke:#1b5e20,stroke-width:2px
    
    class B1,B2,B3 directClass
    class C1,C2,C3 holeClass
    class D1,D2,D3 relayClass
    class E1,E2,E3 protocolClass

import asyncio
import socket
import json
import time
from typing import Tuple, Optional

class UDPHolePuncher:
    """UDP打洞实现类"""
    
    def __init__(self, server_addr: Tuple[str, int]):
        self.server_addr = server_addr
        self.local_socket = None
        self.peer_info = None
        
    async def punch_hole(self, peer_id: str) -> Optional[Tuple[str, int]]:
        """执行UDP打洞流程"""
        try:
            # 1. 创建本地UDP套接字
            self.local_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            self.local_socket.bind(('0.0.0.0', 0))
            local_port = self.local_socket.getsockname()[1]
            
            # 2. 向信令服务器注册
            await self._register_with_server(peer_id, local_port)
            
            # 3. 获取对端信息
            peer_info = await self._get_peer_info(peer_id)
            if not peer_info:
                return None
                
            # 4. 执行同步打洞
            success = await self._synchronized_punch(peer_info)
            
            if success:
                return peer_info['public_addr']
            return None
            
        except Exception as e:
            print(f"UDP打洞失败: {e}")
            return None
    
    async def _register_with_server(self, peer_id: str, local_port: int):
        """向信令服务器注册本地信息"""
        register_msg = {
            'action': 'register',
            'peer_id': peer_id,
            'local_port': local_port,
            'timestamp': time.time()
        }
        
        # 发送注册消息到信令服务器
        msg_bytes = json.dumps(register_msg).encode()
        self.local_socket.sendto(msg_bytes, self.server_addr)
        
        # 等待服务器确认
        response, _ = self.local_socket.recvfrom(1024)
        response_data = json.loads(response.decode())
        
        if response_data.get('status') != 'registered':
            raise Exception("服务器注册失败")
    
    async def _synchronized_punch(self, peer_info: dict) -> bool:
        """执行同步打洞"""
        peer_addr = tuple(peer_info['public_addr'])
        punch_msg = b'PUNCH_HELLO'
        
        # 发送打洞包的策略：
        # 1. 快速发送阶段 - 100ms间隔发送10次
        # 2. 慢速发送阶段 - 500ms间隔发送20次
        # 3. 超时退出
        
        for phase in ['fast', 'slow']:
            interval = 0.1 if phase == 'fast' else 0.5
            count = 10 if phase == 'fast' else 20
            
            for i in range(count):
                try:
                    self.local_socket.sendto(punch_msg, peer_addr)
                    
                    # 非阻塞接收
                    self.local_socket.settimeout(interval)
                    try:
                        data, addr = self.local_socket.recvfrom(1024)
                        if data == b'PUNCH_HELLO' and addr == peer_addr:
                            # 打洞成功，发送确认
                            self.local_socket.sendto(b'PUNCH_ACK', peer_addr)
                            return True
                    except socket.timeout:
                        pass
                        
                    await asyncio.sleep(interval)
                    
                except Exception as e:
                    print(f"打洞发送失败: {e}")
                    
        return False

sequenceDiagram
    participant C1 as 客户端A
    participant N1 as NAT-A
    participant S as 信令服务器
    participant N2 as NAT-B
    participant C2 as 客户端B
    
    Note over C1,C2: Phase 1: 信息收集
    C1->>+S: 注册请求(本地端口)
    S->>N1: 记录公网映射
    N1->>S: 
    S->>-C1: 注册成功
    
    C2->>+S: 注册请求(本地端口)
    S->>N2: 记录公网映射
    N2->>S: 
    S->>-C2: 注册成功
    
    Note over C1,C2: Phase 2: 信息交换
    C1->>+S: 请求B的信息
    S->>-C1: B的公网地址
    
    C2->>+S: 请求A的信息
    S->>-C2: A的公网地址
    
    Note over C1,C2: Phase 3: 同步打洞
    par 同时发送打洞包
        C1->>N1: UDP包 ->> B公网地址
        N1->>N2: 穿越防火墙
        N2->>C2: 到达B
    and
        C2->>N2: UDP包 ->> A公网地址
        N2->>N1: 穿越防火墙
        N1->>C1: 到达A
    end
    
    Note over C1,C2: Phase 4: 连接建立
    C1->>C2: 直接P2P通信
    C2->>C1: 直接P2P通信

import socket
import threading
import time
import select
from typing import Optional, Tuple

class TCPHolePuncher:
    """TCP打洞实现 - 处理复杂的TCP状态管理"""
    
    def __init__(self):
        self.local_socket = None
        self.punch_success = False
        
    def punch_tcp_hole(self, local_port: int, 
                      remote_addr: Tuple[str, int],
                      timeout: int = 30) -> Optional[socket.socket]:
        """执行TCP打洞"""
        
        # 创建本地监听套接字
        listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        listen_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        listen_sock.bind(('0.0.0.0', local_port))
        listen_sock.listen(1)
        listen_sock.settimeout(1)  # 非阻塞监听
        
        # 创建连接套接字
        connect_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        connect_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        connect_sock.bind(('0.0.0.0', local_port))  # 绑定相同端口
        
        start_time = time.time()
        
        try:
            while time.time() - start_time < timeout:
                # 尝试连接
                try:
                    connect_sock.settimeout(0.1)
                    result = connect_sock.connect_ex(remote_addr)
                    
                    if result == 0:  # 连接成功
                        return connect_sock
                    elif result in [115, 36]:  # EINPROGRESS, EALREADY
                        # 连接正在进行中，检查是否完成
                        ready = select.select([], [connect_sock], [], 0.1)
                        if ready[1]:
                            error = connect_sock.getsockopt(socket.SOL_SOCKET, 
                                                          socket.SO_ERROR)
                            if error == 0:
                                return connect_sock
                                
                except socket.error as e:
                    if e.errno not in [115, 36, 111]:  # 忽略预期的错误
                        pass
                
                # 检查是否有入站连接
                try:
                    client_sock, addr = listen_sock.accept()
                    if addr[0] == remote_addr[0]:  # 验证来源
                        listen_sock.close()
                        connect_sock.close()
                        return client_sock
                except socket.timeout:
                    pass
                
                time.sleep(0.05)  # 短暂等待
                
        finally:
            listen_sock.close()
            if connect_sock:
                connect_sock.close()
                
        return None
    
    def enhanced_tcp_punch(self, peer_info: dict) -> Optional[socket.socket]:
        """增强型TCP打洞 - 支持多种策略"""
        strategies = [
            self._strategy_simultaneous_open,
            self._strategy_sequential_connect,
            self._strategy_port_prediction
        ]
        
        for strategy in strategies:
            try:
                result = strategy(peer_info)
                if result:
                    return result
            except Exception as e:
                print(f"策略失败: {strategy.__name__}, 错误: {e}")
                continue
                
        return None
    
    def _strategy_port_prediction(self, peer_info: dict) -> Optional[socket.socket]:
        """端口预测策略 - 针对端口分配规律的NAT"""
        base_port = peer_info['public_addr'][1]
        
        # 尝试预测的端口范围
        predicted_ports = []
        
        # 线性递增预测
        for i in range(1, 10):
            predicted_ports.append(base_port + i)
            predicted_ports.append(base_port - i)
            
        # 常见的端口分配模式
        predicted_ports.extend([
            base_port + 2,   # 某些NAT的步长为2
            base_port + 4,   # 某些NAT的步长为4
            base_port | 1,   # 奇偶端口切换
            base_port & ~1   # 偶数端口对齐
        ])
        
        for port in predicted_ports:
            if 1024 <= port <= 65535:
                try:
                    target_addr = (peer_info['public_addr'][0], port)
                    sock = self.punch_tcp_hole(
                        peer_info['local_port'], 
                        target_addr, 
                        timeout=5
                    )
                    if sock:
                        return sock
                except Exception:
                    continue
                    
        return None

class HybridNATTraversal:
    """混合穿透策略 - 结合多种技术的智能穿透"""
    
    def __init__(self):
        self.udp_puncher = UDPHolePuncher(('stun.server.com', 3478))
        self.tcp_puncher = TCPHolePuncher()
        self.relay_client = None
        
    async def establish_connection(self, peer_id: str) -> dict:
        """建立连接的完整流程"""
        connection_info = {
            'method': None,
            'socket': None,
            'latency': None,
            'bandwidth': None
        }
        
        # 策略1: 尝试UDP打洞 (最快)
        print("尝试UDP打洞...")
        udp_result = await self.udp_puncher.punch_hole(peer_id)
        if udp_result:
            connection_info.update({
                'method': 'UDP_PUNCH',
                'socket': self.udp_puncher.local_socket,
                'latency': await self._measure_latency(udp_result)
            })
            return connection_info
        
        # 策略2: 尝试TCP打洞 (中等延迟)
        print("UDP打洞失败，尝试TCP打洞...")
        peer_info = await self._get_peer_tcp_info(peer_id)
        if peer_info:
            tcp_sock = self.tcp_puncher.enhanced_tcp_punch(peer_info)
            if tcp_sock:
                connection_info.update({
                    'method': 'TCP_PUNCH',
                    'socket': tcp_sock,
                    'latency': await self._measure_tcp_latency(tcp_sock)
                })
                return connection_info
        
        # 策略3: 回退到中继模式 (最高延迟但保证连通)
        print("直接打洞失败，使用中继模式...")
        relay_conn = await self._establish_relay_connection(peer_id)
        if relay_conn:
            connection_info.update({
                'method': 'RELAY',
                'socket': relay_conn,
                'latency': await self._measure_relay_latency(relay_conn)
            })
            return connection_info
        
        # 所有策略都失败
        connection_info['method'] = 'FAILED'
        return connection_info
    
    async def _measure_latency(self, peer_addr: Tuple[str, int]) -> float:
        """测量UDP连接延迟"""
        start_time = time.time()
        
        # 发送ping包
        ping_msg = b'PING_' + str(int(start_time * 1000)).encode()
        self.udp_puncher.local_socket.sendto(ping_msg, peer_addr)
        
        # 等待pong响应
        try:
            self.udp_puncher.local_socket.settimeout(2)
            response, addr = self.udp_puncher.local_socket.recvfrom(1024)
            
            if response.startswith(b'PONG_') and addr == peer_addr:
                return (time.time() - start_time) * 1000  # 返回毫秒
        except socket.timeout:
            pass
            
        return float('inf')  # 超时返回无穷大

// WebRTC P2P连接实现
class WebRTCNATTraversal {
    constructor(signalingServer) {
        this.signalingServer = signalingServer;
        this.peerConnection = null;
        this.dataChannel = null;
        this.localStream = null;
        
        // ICE服务器配置
        this.iceServers = [
            { urls: 'stun:stun.l.google.com:19302' },
            { urls: 'stun:stun1.l.google.com:19302' },
            {
                urls: 'turn:turnserver.com:3478',
                username: 'user',
                credential: 'pass'
            }
        ];
    }
    
    async initializePeerConnection() {
        // 创建RTCPeerConnection
        this.peerConnection = new RTCPeerConnection({
            iceServers: this.iceServers,
            iceCandidatePoolSize: 10
        });
        
        // 设置事件监听器
        this.peerConnection.onicecandidate = (event) => {
            if (event.candidate) {
                this.signalingServer.send({
                    type: 'ice-candidate',
                    candidate: event.candidate
                });
            }
        };
        
        this.peerConnection.onconnectionstatechange = () => {
            console.log('连接状态:', this.peerConnection.connectionState);
        };
        
        // 创建数据通道
        this.dataChannel = this.peerConnection.createDataChannel('data', {
            ordered: true,
            maxRetransmits: 3
        });
        
        this.dataChannel.onopen = () => {
            console.log('数据通道已打开');
            this.onDataChannelOpen();
        };
        
        this.dataChannel.onmessage = (event) => {
            this.onDataReceived(event.data);
        };
    }
    
    async createOffer() {
        const offer = await this.peerConnection.createOffer();
        await this.peerConnection.setLocalDescription(offer);
        
        this.signalingServer.send({
            type: 'offer',
            sdp: offer
        });
    }
    
    async handleOffer(offer) {
        await this.peerConnection.setRemoteDescription(offer);
        
        const answer = await this.peerConnection.createAnswer();
        await this.peerConnection.setLocalDescription(answer);
        
        this.signalingServer.send({
            type: 'answer',
            sdp: answer
        });
    }
    
    async handleAnswer(answer) {
        await this.peerConnection.setRemoteDescription(answer);
    }
    
    async handleIceCandidate(candidate) {
        await this.peerConnection.addIceCandidate(candidate);
    }
    
    sendData(data) {
        if (this.dataChannel && this.dataChannel.readyState === 'open') {
            this.dataChannel.send(data);
        }
    }
    
    // 连接质量监控
    async getConnectionStats() {
        const stats = await this.peerConnection.getStats();
        const connectionStats = {};
        
        stats.forEach((report) => {
            if (report.type === 'candidate-pair' && report.state === 'succeeded') {
                connectionStats.rtt = report.currentRoundTripTime * 1000; // 转换为毫秒
                connectionStats.bytesReceived = report.bytesReceived;
                connectionStats.bytesSent = report.bytesSent;
            }
        });
        
        return connectionStats;
    }
}

import hashlib
import hmac
import time
import secrets
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64

class SecureNATTraversal:
    """安全的内网穿透实现"""
    
    def __init__(self, shared_secret: str):
        self.shared_secret = shared_secret.encode()
        self.session_key = None
        self.cipher_suite = None
        
    def generate_session_key(self, salt: bytes = None) -> bytes:
        """生成会话密钥"""
        if salt is None:
            salt = secrets.token_bytes(16)
            
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=100000,
        )
        
        self.session_key = kdf.derive(self.shared_secret)
        self.cipher_suite = Fernet(base64.urlsafe_b64encode(self.session_key))
        return salt
    
    def create_authenticated_message(self, message: bytes, 
                                   timestamp: float = None) -> dict:
        """创建认证消息"""
        if timestamp is None:
            timestamp = time.time()
            
        # 添加时间戳防重放攻击
        timestamped_msg = f"{timestamp}:{message.decode()}".encode()
        
        # 加密消息
        encrypted_msg = self.cipher_suite.encrypt(timestamped_msg)
        
        # 生成HMAC签名
        signature = hmac.new(
            self.shared_secret,
            encrypted_msg,
            hashlib.sha256
        ).hexdigest()
        
        return {
            'encrypted_data': base64.b64encode(encrypted_msg).decode(),
            'signature': signature,
            'timestamp': timestamp
        }
    
    def verify_and_decrypt_message(self, auth_message: dict, 
                                 max_age: int = 300) -> bytes:
        """验证并解密消息"""
        try:
            # 检查时间戳
            current_time = time.time()
            if current_time - auth_message['timestamp'] > max_age:
                raise ValueError("消息已过期")
            
            # 验证签名
            encrypted_data = base64.b64decode(auth_message['encrypted_data'])
            expected_signature = hmac.new(
                self.shared_secret,
                encrypted_data,
                hashlib.sha256
            ).hexdigest()
            
            if not hmac.compare_digest(expected_signature, auth_message['signature']):
                raise ValueError("签名验证失败")
            
            # 解密消息
            decrypted_data = self.cipher_suite.decrypt(encrypted_data)
            
            # 提取原始消息（去除时间戳）
            timestamp_str, original_msg = decrypted_data.decode().split(':', 1)
            
            return original_msg.encode()
            
        except Exception as e:
            raise ValueError(f"消息验证失败: {e}")
    
    def secure_hole_punch(self, peer_addr: tuple, local_socket: socket.socket):
        """安全的打洞实现"""
        # 生成随机nonce
        nonce = secrets.token_bytes(16)
        
        # 创建握手消息
        handshake_data = {
            'type': 'handshake',
            'nonce': base64.b64encode(nonce).decode(),
            'protocol_version': '1.0'
        }
        
        # 发送认证的握手消息
        auth_handshake = self.create_authenticated_message(
            json.dumps(handshake_data).encode()
        )
        
        local_socket.sendto(
            json.dumps(auth_handshake).encode(),
            peer_addr
        )
        
        # 等待响应并验证
        try:
            response_data, addr = local_socket.recvfrom(2048)
            if addr != peer_addr:
                raise ValueError("响应来源地址不匹配")
                
            response_auth = json.loads(response_data.decode())
            decrypted_response = self.verify_and_decrypt_message(response_auth)
            
            response_obj = json.loads(decrypted_response.decode())
            
            # 验证nonce防重放
            if response_obj.get('nonce') != base64.b64encode(nonce).decode():
                raise ValueError("Nonce验证失败")
                
            return True
            
        except Exception as e:
            print(f"安全握手失败: {e}")
            return False

import asyncio
import websockets
import json
import logging
from typing import Dict, Set
import redis
import uuid

class EnterpriseNATTraversalServer:
    """企业级内网穿透服务器"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.clients: Dict[str, websockets.WebSocketServerProtocol] = {}
        self.peer_registry: Dict[str, dict] = {}
        self.redis_client = redis.from_url(redis_url)
        self.logger = logging.getLogger(__name__)
        
    async def start_server(self, host: str = "0.0.0.0", port: int = 8765):
        """启动WebSocket信令服务器"""
        self.logger.info(f"启动信令服务器 {host}:{port}")
        
        async with websockets.serve(
            self.handle_client,
            host,
            port,
            ping_interval=30,
            ping_timeout=10
        ):
            await asyncio.Future()  # 永久运行
    
    async def handle_client(self, websocket, path):
        """处理客户端连接"""
        client_id = str(uuid.uuid4())
        self.clients[client_id] = websocket
        
        try:
            self.logger.info(f"客户端连接: {client_id}")
            
            async for message in websocket:
                try:
                    data = json.loads(message)
                    await self.process_message(client_id, data)
                except json.JSONDecodeError:
                    await self.send_error(websocket, "无效的JSON格式")
                except Exception as e:
                    self.logger.error(f"处理消息错误: {e}")
                    
        except websockets.exceptions.ConnectionClosed:
            self.logger.info(f"客户端断开: {client_id}")
        finally:
            await self.cleanup_client(client_id)
    
    async def process_message(self, client_id: str, data: dict):
        """处理客户端消息"""
        message_type = data.get('type')
        
        handlers = {
            'register': self.handle_register,
            'find_peer': self.handle_find_peer,
            'offer': self.handle_offer,
            'answer': self.handle_answer,
            'ice_candidate': self.handle_ice_candidate,
            'heartbeat': self.handle_heartbeat
        }
        
        handler = handlers.get(message_type)
        if handler:
            await handler(client_id, data)
        else:
            await self.send_error(
                self.clients[client_id],
                f"未知消息类型: {message_type}"
            )
    
    async def handle_register(self, client_id: str, data: dict):
        """处理客户端注册"""
        peer_info = {
            'client_id': client_id,
            'peer_name': data.get('peer_name'),
            'capabilities': data.get('capabilities', []),
            'public_ip': data.get('public_ip'),
            'local_ports': data.get('local_ports', []),
            'nat_type': data.get('nat_type'),
            'timestamp': asyncio.get_event_loop().time()
        }
        
        # 存储到Redis
        await self.redis_client.hset(
            f"peer:{client_id}",
            mapping=peer_info
        )
        
        # 设置过期时间
        await self.redis_client.expire(f"peer:{client_id}", 300)
        
        self.peer_registry[client_id] = peer_info
        
        await self.send_message(self.clients[client_id], {
            'type': 'register_success',
            'client_id': client_id
        })
        
        self.logger.info(f"客户端注册成功: {peer_info['peer_name']}")
    
    async def handle_find_peer(self, client_id: str, data: dict):
        """处理查找对等端请求"""
        target_name = data.get('target_name')
        
        # 从Redis查找目标客户端
        peer_keys = await self.redis_client.keys("peer:*")
        target_peer = None
        
        for key in peer_keys:
            peer_data = await self.redis_client.hgetall(key)
            if peer_data.get('peer_name') == target_name:
                target_peer = peer_data
                break
        
        if target_peer:
            # 返回目标客户端信息（去除敏感信息）
            safe_peer_info = {
                'client_id': target_peer['client_id'],
                'peer_name': target_peer['peer_name'],
                'public_ip': target_peer['public_ip'],
                'nat_type': target_peer['nat_type'],
                'capabilities': target_peer['capabilities']
            }
            
            await self.send_message(self.clients[client_id], {
                'type': 'peer_found',
                'peer_info': safe_peer_info
            })
        else:
            await self.send_message(self.clients[client_id], {
                'type': 'peer_not_found',
                'target_name': target_name
            })
    
    async def handle_offer(self, client_id: str, data: dict):
        """处理WebRTC Offer"""
        target_id = data.get('target_id')
        
        if target_id in self.clients:
            # 转发offer到目标客户端
            await self.send_message(self.clients[target_id], {
                'type': 'offer',
                'from_id': client_id,
                'sdp': data.get('sdp')
            })
        else:
            await self.send_error(
                self.clients[client_id],
                f"目标客户端不在线: {target_id}"
            )
    
    async def send_message(self, websocket, message: dict):
        """发送消息到客户端"""
        try:
            await websocket.send(json.dumps(message))
        except websockets.exceptions.ConnectionClosed:
            pass
    
    async def send_error(self, websocket, error_message: str):
        """发送错误消息"""
        await self.send_message(websocket, {
            'type': 'error',
            'message': error_message
        })
    
    async def cleanup_client(self, client_id: str):
        """清理客户端资源"""
        if client_id in self.clients:
            del self.clients[client_id]
        
        if client_id in self.peer_registry:
            del self.peer_registry[client_id]
        
        # 从Redis删除
        await self.redis_client.delete(f"peer:{client_id}")
        
        self.logger.info(f"客户端资源清理完成: {client_id}")

# 启动服务器
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    server = EnterpriseNATTraversalServer()
    asyncio.run(server.start_server())

import asyncio
import time
import statistics
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class ConnectionMetrics:
    """连接质量指标"""
    latency: float  # 延迟(ms)
    jitter: float   # 抖动(ms)
    packet_loss: float  # 丢包率(%)
    bandwidth: float    # 带宽(Mbps)
    connection_time: float  # 连接建立时间(ms)

class NATTraversalMonitor:
    """内网穿透监控系统"""
    
    def __init__(self):
        self.metrics_history: List[ConnectionMetrics] = []
        self.active_connections: dict = {}
        
    async def monitor_connection_quality(self, connection_id: str, 
                                       socket_obj) -> ConnectionMetrics:
        """监控连接质量"""
        # 测量延迟和抖动
        latencies = await self._measure_latencies(socket_obj, samples=10)
        avg_latency = statistics.mean(latencies)
        jitter = statistics.stdev(latencies) if len(latencies) > 1 else 0
        
        # 测量丢包率
        packet_loss = await self._measure_packet_loss(socket_obj)
        
        # 测量带宽
        bandwidth = await self._measure_bandwidth(socket_obj)
        
        metrics = ConnectionMetrics(
            latency=avg_latency,
            jitter=jitter,
            packet_loss=packet_loss,
            bandwidth=bandwidth,
            connection_time=0  # 在连接建立时设置
        )
        
        self.metrics_history.append(metrics)
        return metrics
    
    async def _measure_latencies(self, socket_obj, samples: int = 10) -> List[float]:
        """测量延迟样本"""
        latencies = []
        
        for i in range(samples):
            start_time = time.perf_counter()
            
            # 发送ping包
            ping_data = f"PING_{i}_{start_time}".encode()
            socket_obj.send(ping_data)
            
            # 等待pong响应
            try:
                socket_obj.settimeout(2.0)
                response = socket_obj.recv(1024)
                
                if response.startswith(b'PONG_'):
                    end_time = time.perf_counter()
                    latency = (end_time - start_time) * 1000  # 转换为毫秒
                    latencies.append(latency)
                    
            except socket.timeout:
                latencies.append(2000)  # 超时记为2秒
            
            await asyncio.sleep(0.1)  # 间隔100ms
        
        return latencies
    
    def generate_quality_report(self) -> dict:
        """生成连接质量报告"""
        if not self.metrics_history:
            return {"error": "无监控数据"}
        
        recent_metrics = self.metrics_history[-100:]  # 最近100个样本
        
        return {
            "average_latency": statistics.mean([m.latency for m in recent_metrics]),
            "max_latency": max([m.latency for m in recent_metrics]),
            "min_latency": min([m.latency for m in recent_metrics]),
            "average_jitter": statistics.mean([m.jitter for m in recent_metrics]),
            "average_packet_loss": statistics.mean([m.packet_loss for m in recent_metrics]),
            "average_bandwidth": statistics.mean([m.bandwidth for m in recent_metrics]),
            "connection_stability": self._calculate_stability_score(recent_metrics),
            "total_samples": len(recent_metrics)
        }
    
    def _calculate_stability_score(self, metrics: List[ConnectionMetrics]) -> float:
        """计算连接稳定性评分 (0-100)"""
        if not metrics:
            return 0
        
        # 基于延迟稳定性、丢包率、抖动计算综合评分
        latency_stability = 100 - min(100, statistics.stdev([m.latency for m in metrics]) / 10)
        packet_loss_penalty = statistics.mean([m.packet_loss for m in metrics]) * 2
        jitter_penalty = statistics.mean([m.jitter for m in metrics]) / 5
        
        stability_score = max(0, latency_stability - packet_loss_penalty - jitter_penalty)
        return round(stability_score, 2)

pie title "穿透方案使用分布"
    "UDP打洞" : 45
    "TCP打洞" : 25
    "WebRTC" : 20
    "中继转发" : 10