// Package corebind 是 Meshray-Manager 侧的 wireguard-go 集成层(形态 B:增强传输)。 // // 它实现 wireguard-go 的 conn.Bind 接口(EnhancedBind),完全接管 WG 的 UDP 收发, // 把密文流量导向 meshray 核心(git.zkcoi.com/zkcoi/meshray)的 9 层传输。 // 该层属于「管理器 / UI」范畴,而非核心;核心本身零 wireguard 依赖。 package corebind import ( "net" "sync" "git.zkcoi.com/zkcoi/meshray/engine" "golang.zx2c4.com/wireguard/conn" "go.uber.org/zap" ) // EnhancedBind 是 corebind 提供的 conn.Bind 实现,完全接管 WireGuard 的 UDP 收发, // 使所有 WG 密文流量都经由 meshray 核心的 9 层传输(Direct/TURN/WebRTC/...)而非裸 UDP。 // // 设计要点: // - Direct-UDP 由一个绑定在 listen_port 的*共享未连接 UDP socket*(directSock)承载, // 从任意来源收包,把「真实源地址」作为 Endpoint 交给 WG,WG 用密文内的 receiver index // 匹配 peer 并据此更新对端出向地址(学址)。由此对称直连与 NAT 穿透学址原生可通。 // - TURN/WebRTC/WS 等中继层经核心传输连接(connMgr)承载,Direct 不可用时补充。 // - 集成通过核心暴露的钩子接入:SetPeerRegistrar / SetRelayInboundHandler / // SetDirectByBind,核心永不反向依赖本包。 type EnhancedBind struct { inner conn.Bind networkID string engine *core.Engine logger *zap.Logger // listenPort 本端 WG 监听端口;Direct-UDP 共享监听 socket 绑定于此,形成对称直连。 listenPort int mu sync.RWMutex peerByAddr map[string]string // endpoint addr(str) -> peerKey(Send 反查中继连接) peerEndpoint map[string]conn.Endpoint // peerKey -> 对端真实 Endpoint(中继入站回灌用) // directSock 承载 Direct-UDP 的共享未连接 UDP socket(绑定 listenPort)。 // 从任意来源收包(学习对端经 NAT 后的真实源地址),并按目标地址发包。 directSock *net.UDPConn inboundCh chan inboundPacket closed bool } // inboundPacket 入站密文;携带来源信息以便回灌 WG 时定位对端 Endpoint。 type inboundPacket struct { peerKey string // 中继入站时按 peerKey 查 Endpoint ep conn.Endpoint // Direct 共享 socket 学得的真实源地址;非空时优先使用 data []byte } // NewEnhancedBind 创建增强 Bind,并关联指定组网的 Engine 单例; // 同时经核心钩子把入站回灌接到本 Bind,使 9 层传输连接读到的包能回到 WG。 func NewEnhancedBind(listenPort int, networkID string, logger *zap.Logger) (*EnhancedBind, error) { eng := getOrCreateEngine(networkID, logger) inner := conn.NewDefaultBind() b := &EnhancedBind{ inner: inner, networkID: networkID, engine: eng, logger: logger, listenPort: listenPort, peerByAddr: make(map[string]string), peerEndpoint: make(map[string]conn.Endpoint), inboundCh: make(chan inboundPacket, 1024), } // 通过核心钩子接入,避免核心反向依赖本包: eng.SetPeerRegistrar(b) // 收到对端候选时注册 endpoint→peerKey eng.SetRelayInboundHandler(b.deliverInbound) // 传输连接入站回灌 WG eng.SetDirectByBind(true) // Direct 由本 Bind 共享 socket 承载 eng.SetListenPort(listenPort) return b, nil } // Engine 返回关联的引擎实例(standalone 二进制用于注入候选触发拨号) func (b *EnhancedBind) Engine() *core.Engine { return b.engine } // RegisterPeer 注册 endpoint 地址到 peerKey 的映射,并缓存对端真实 Endpoint 供回灌使用。 // 由 Engine.NotifyPeerInfo 在收到候选时自动调用;standalone 也可直接调用。 func (b *EnhancedBind) RegisterPeer(addr string, peerKey string) { ep, err := b.inner.ParseEndpoint(addr) if err != nil { b.logger.Warn("解析对端 Endpoint 失败,回灌将使用 0.0.0.0:0", zap.String("addr", addr), zap.Error(err)) ep = nil } b.mu.Lock() b.peerByAddr[addr] = peerKey if ep != nil { b.peerEndpoint[peerKey] = ep } b.mu.Unlock() } // Open 创建 Direct-UDP 共享监听 socket(绑定 listen_port),并启动接收协程。 // 返回从 inboundCh 读取入站密文的接收函数:Direct 入站与中继入站统一经此回灌 WG。 func (b *EnhancedBind) Open(port uint16) ([]conn.ReceiveFunc, uint16, error) { b.mu.Lock() b.closed = false // 端口优先用 WG 传入的 port,其次用构造时的 listenPort;0 表示由 OS 分配。 bindPort := int(port) if bindPort == 0 { bindPort = b.listenPort } // 关闭可能残留的旧 socket(重开场景) if b.directSock != nil { b.directSock.Close() b.directSock = nil } b.mu.Unlock() sock, err := net.ListenUDP("udp", &net.UDPAddr{Port: bindPort}) if err != nil { return nil, 0, err } actual := sock.LocalAddr().(*net.UDPAddr).Port b.mu.Lock() b.directSock = sock b.listenPort = actual b.mu.Unlock() // 回填引擎实际监听端口(listen_port=0 自动分配时同步) if b.engine != nil { b.engine.SetListenPort(actual) } go b.readDirectLoop(sock) b.logger.Info("Direct-UDP 共享监听 socket 已就绪", zap.Int("listen_port", actual)) // v4/v6 各一个接收函数,均从同一个 inboundCh 读取 return []conn.ReceiveFunc{b.receiveFunc, b.receiveFunc}, uint16(actual), nil } // readDirectLoop 从共享监听 socket 收包(任意来源),把「真实源地址」作为 Endpoint 投入 inboundCh。 // 这是 NAT 穿透学址的核心:公网侧据此学到对端 NAT 映射后的真实地址,回包得以正确返回。 func (b *EnhancedBind) readDirectLoop(sock *net.UDPConn) { buf := make([]byte, 65535) for { n, src, err := sock.ReadFromUDP(buf) if err != nil { b.mu.RLock() closed := b.closed b.mu.RUnlock() if closed { return } b.logger.Debug("Direct 共享 socket 读取失败,停止接收", zap.Error(err)) return } data := make([]byte, n) copy(data, buf[:n]) ep, perr := b.inner.ParseEndpoint(src.String()) if perr != nil { b.logger.Debug("解析入站源地址失败,丢弃", zap.String("src", src.String()), zap.Error(perr)) continue } select { case b.inboundCh <- inboundPacket{ep: ep, data: data}: default: b.logger.Warn("入站队列满,丢弃包", zap.String("src", src.String())) } } } // receiveFunc 从 inboundCh 取出入站密文,拷贝进 WG 提供的缓冲(保持 WG 缓冲所有权), // 并填入来源 Endpoint 供 WG 握手更新出向地址: // - Direct 入站:ep 为共享 socket 学得的真实源地址(优先); // - 中继入站:ep 为空,按 peerKey 查缓存的对端 Endpoint。 func (b *EnhancedBind) receiveFunc(packets [][]byte, sizes []int, eps []conn.Endpoint) (int, error) { first, ok := <-b.inboundCh if !ok { return 0, net.ErrClosed } n := 0 copy(packets[n], first.data) sizes[n] = len(first.data) eps[n] = b.endpointForPacket(first) n++ for n < len(packets) { select { case pkt := <-b.inboundCh: copy(packets[n], pkt.data) sizes[n] = len(pkt.data) eps[n] = b.endpointForPacket(pkt) n++ default: return n, nil } } return n, nil } // endpointForPacket 返回入站包的来源 Endpoint:Direct 入站直接用其学得的 ep; // 中继入站按 peerKey 查缓存,未知时退化为 0.0.0.0:0。 func (b *EnhancedBind) endpointForPacket(pkt inboundPacket) conn.Endpoint { if pkt.ep != nil { return pkt.ep } b.mu.RLock() defer b.mu.RUnlock() if ep, ok := b.peerEndpoint[pkt.peerKey]; ok { return ep } ep, _ := b.inner.ParseEndpoint("0.0.0.0:0") return ep } // Send 出站路径: // - 若该对端已建立中继/信令层连接(TURN/WebRTC/WS,位于 connMgr),经其发送; // - 否则(Direct-UDP)经共享监听 socket 直接发往目标地址(WG 维护的对端 Endpoint: // 初始为配置 endpoint,学址后为对端真实地址)。 func (b *EnhancedBind) Send(bufs [][]byte, ep conn.Endpoint) error { addr := ep.DstToString() b.mu.RLock() peerKey, mapped := b.peerByAddr[addr] b.mu.RUnlock() if mapped { if c, ok := b.engine.GetConnMgr().Get(peerKey); ok && c != nil { for _, buf := range bufs { if len(buf) == 0 { continue } if _, err := c.Write(buf); err != nil { b.logger.Debug("经中继连接发送失败,转 Direct 共享 socket", zap.String("peer", peerKey), zap.Error(err)) return b.directSendTo(addr, bufs) } } return nil } } // Direct-UDP:共享监听 socket 直发(NAT 穿透原生支持) return b.directSendTo(addr, bufs) } // directSendTo 经共享监听 socket 把数据发往目标地址。源端口固定为 listen_port, // 满足对称直连;NAT 后主动侧发包后,NAT 记录会话,公网侧回包得以返回。 func (b *EnhancedBind) directSendTo(addr string, bufs [][]byte) error { b.mu.RLock() sock := b.directSock b.mu.RUnlock() if sock == nil { return net.ErrClosed } udpAddr, err := net.ResolveUDPAddr("udp", addr) if err != nil { return err } for _, buf := range bufs { if len(buf) == 0 { continue } if _, err := sock.WriteToUDP(buf, udpAddr); err != nil { return err } } return nil } // deliverInbound 由引擎中继传输连接读取到入站时调用,把密文投入 inboundCh 回灌 WG。 // 中继入站不携带来源地址,按 peerKey 在回灌时查对端 Endpoint。 func (b *EnhancedBind) deliverInbound(peerKey string, packet []byte) { cp := make([]byte, len(packet)) copy(cp, packet) select { case b.inboundCh <- inboundPacket{peerKey: peerKey, data: cp}: default: b.logger.Warn("入站队列满,丢弃包", zap.String("peer", peerKey)) } } // Close 关闭共享监听 socket 与入站通道(使接收函数返回 net.ErrClosed)。 func (b *EnhancedBind) Close() error { b.mu.Lock() if !b.closed { b.closed = true close(b.inboundCh) } if b.directSock != nil { b.directSock.Close() b.directSock = nil } b.mu.Unlock() return b.inner.Close() } // SetMark 本实现不涉及内核 socket,置空标记为 no-op func (b *EnhancedBind) SetMark(mark uint32) error { return nil } // ParseEndpoint 委托默认 Bind 解析 func (b *EnhancedBind) ParseEndpoint(s string) (conn.Endpoint, error) { return b.inner.ParseEndpoint(s) } // BatchSize 返回默认 Bind 的批量大小 func (b *EnhancedBind) BatchSize() int { return b.inner.BatchSize() }