package openp2p import ( "bytes" "encoding/binary" "encoding/json" "fmt" "math/rand" "net" "time" ) func handshakeC2C(t *P2PTunnel) (err error) { gLog.d("handshakeC2C %s:%d:%d to %s:%d", gConf.Network.Node, t.coneLocalPort, t.coneNatPort, t.config.peerIP, t.config.peerConeNatPort) defer gLog.d("handshakeC2C end") conn, err := net.ListenUDP("udp", t.localHoleAddr) if err != nil { return err } defer conn.Close() _, err = UDPWrite(conn, t.remoteHoleAddr, MsgP2P, MsgPunchHandshake, P2PHandshakeReq{ID: t.id}) if err != nil { gLog.d("handshakeC2C write MsgPunchHandshake error:%s", err) return err } ra, head, buff, _, err := UDPRead(conn, HandshakeTimeout) if err != nil { gLog.d("handshakeC2C read MsgPunchHandshake error:%s", err) return err } t.remoteHoleAddr, _ = net.ResolveUDPAddr("udp", ra.String()) var tunnelID uint64 if len(buff) > openP2PHeaderSize { req := P2PHandshakeReq{} if err := json.Unmarshal(buff[openP2PHeaderSize:openP2PHeaderSize+int(head.DataLen)], &req); err == nil { tunnelID = req.ID } } else { // compatible with old version tunnelID = t.id } if head.MainType == MsgP2P && head.SubType == MsgPunchHandshake && tunnelID == t.id { gLog.d("read tunnelid:%d handshake ", t.id) UDPWrite(conn, t.remoteHoleAddr, MsgP2P, MsgPunchHandshakeAck, P2PHandshakeReq{ID: t.id}) _, head, _, _, err = UDPRead(conn, HandshakeTimeout) if err != nil { gLog.d("handshakeC2C write MsgPunchHandshakeAck error:", err) return err } } if head.MainType == MsgP2P && head.SubType == MsgPunchHandshakeAck && tunnelID == t.id { gLog.d("read tunnelID:%d handshake ack ", t.id) _, err = UDPWrite(conn, t.remoteHoleAddr, MsgP2P, MsgPunchHandshakeAck, P2PHandshakeReq{ID: t.id}) if err != nil { gLog.d("handshakeC2C write MsgPunchHandshakeAck error:%s", err) return err } } gLog.i("handshakeC2C ok") return nil } func handshakeC2S(t *P2PTunnel) error { gLog.d("tid:%d handshakeC2S start", t.id) defer gLog.d("tid:%d handshakeC2S end", t.id) if !buildTunnelMtx.TryLock() { // time.Sleep(time.Second * 3) return ErrBuildTunnelBusy } defer buildTunnelMtx.Unlock() startTime := time.Now() r := rand.New(rand.NewSource(time.Now().UnixNano())) randPorts := r.Perm(65532) conn, err := net.ListenUDP("udp", t.localHoleAddr) if err != nil { return err } defer conn.Close() go func() error { gLog.d("tid:%d send symmetric handshake to %s from %d:%d start", t.id, t.config.peerIP, t.coneLocalPort, t.coneNatPort) for i := 0; i < SymmetricHandshakeNum; i++ { // time.Sleep(SymmetricHandshakeInterval) dst, err := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", t.config.peerIP, randPorts[i]+2)) if err != nil { return err } _, err = UDPWrite(conn, dst, MsgP2P, MsgPunchHandshake, P2PHandshakeReq{ID: t.id}) if err != nil { gLog.d("tid:%d handshakeC2S write MsgPunchHandshake error:%s", t.id, err) return err } } gLog.d("tid:%d send symmetric handshake end", t.id) return nil }() err = conn.SetReadDeadline(time.Now().Add(HandshakeTimeout)) if err != nil { gLog.d("tid:%d SymmetricHandshakeAckTimeout SetReadDeadline error", t.id) return err } // read response of the punching hole ok port buff := make([]byte, 1024) _, dst, err := conn.ReadFrom(buff) if err != nil { gLog.d("tid:%d handshakeC2S wait timeout", t.id) return err } head := &openP2PHeader{} err = binary.Read(bytes.NewReader(buff[:openP2PHeaderSize]), binary.LittleEndian, head) if err != nil { gLog.e("tid:%d parse p2pheader error:%s", t.id, err) return err } t.remoteHoleAddr, _ = net.ResolveUDPAddr("udp", dst.String()) var tunnelID uint64 if len(buff) > openP2PHeaderSize { req := P2PHandshakeReq{} if err := json.Unmarshal(buff[openP2PHeaderSize:openP2PHeaderSize+int(head.DataLen)], &req); err == nil { tunnelID = req.ID } } else { // compatible with old version tunnelID = t.id } if head.MainType == MsgP2P && head.SubType == MsgPunchHandshake && tunnelID == t.id { gLog.d("tid:%d handshakeC2S read handshake ", t.id) UDPWrite(conn, t.remoteHoleAddr, MsgP2P, MsgPunchHandshakeAck, P2PHandshakeReq{ID: t.id}) for { _, head, buff, _, err = UDPRead(conn, HandshakeTimeout) if err != nil { gLog.d("tid:%d handshakeC2S handshake error", t.id) return err } var tunnelID uint64 if len(buff) > openP2PHeaderSize { req := P2PHandshakeReq{} if err := json.Unmarshal(buff[openP2PHeaderSize:openP2PHeaderSize+int(head.DataLen)], &req); err == nil { tunnelID = req.ID } } else { // compatible with old version tunnelID = t.id } // waiting ack if head.MainType == MsgP2P && head.SubType == MsgPunchHandshakeAck && tunnelID == t.id { break } } } if head.MainType == MsgP2P && head.SubType == MsgPunchHandshakeAck { gLog.d("tid:%d handshakeC2S read handshake ack %s", t.id, t.remoteHoleAddr.String()) _, err = UDPWrite(conn, t.remoteHoleAddr, MsgP2P, MsgPunchHandshakeAck, P2PHandshakeReq{ID: t.id}) return err } else { gLog.d("tid:%d handshakeS2C read msg but not MsgPunchHandshakeAck", t.id) } gLog.i("tid:%d handshakeC2S ok. cost %d ms", t.id, time.Since(startTime)/time.Millisecond) return nil } func handshakeS2C(t *P2PTunnel) error { gLog.d("tid:%d handshakeS2C start", t.id) defer gLog.d("tid:%d handshakeS2C end", t.id) if !buildTunnelMtx.TryLock() { // time.Sleep(time.Second * 3) return ErrBuildTunnelBusy } defer buildTunnelMtx.Unlock() startTime := time.Now() gotCh := make(chan *net.UDPAddr, 50) // sequencely udp send handshake, do not parallel send gLog.d("tid:%d send symmetric handshake to %s:%d start", t.id, t.config.peerIP, t.config.peerConeNatPort) gotIt := false for i := 0; i < SymmetricHandshakeNum; i++ { // time.Sleep(SymmetricHandshakeInterval) go func(t *P2PTunnel) error { conn, err := net.ListenUDP("udp", nil) // TODO: system allocated port really random? if err != nil { gLog.d("tid:%d listen error", t.id) return err } defer conn.Close() UDPWrite(conn, t.remoteHoleAddr, MsgP2P, MsgPunchHandshake, P2PHandshakeReq{ID: t.id}) _, head, buff, _, err := UDPRead(conn, HandshakeTimeout) if err != nil { // gLog.Println(LevelDEBUG, "one of the handshake error:", err) return err } if gotIt { return nil } var tunnelID uint64 if len(buff) >= openP2PHeaderSize+8 { req := P2PHandshakeReq{} if err := json.Unmarshal(buff[openP2PHeaderSize:openP2PHeaderSize+int(head.DataLen)], &req); err == nil { tunnelID = req.ID } } else { // compatible with old version tunnelID = t.id } if head.MainType == MsgP2P && head.SubType == MsgPunchHandshake && tunnelID == t.id { gLog.d("tid:%d handshakeS2C read handshake ", t.id) UDPWrite(conn, t.remoteHoleAddr, MsgP2P, MsgPunchHandshakeAck, P2PHandshakeReq{ID: t.id}) // may read several MsgPunchHandshake for { _, head, buff, _, err = UDPRead(conn, HandshakeTimeout) if err != nil { gLog.d("tid:%d handshakeS2C handshake error", t.id) return err } if len(buff) > openP2PHeaderSize { req := P2PHandshakeReq{} if err := json.Unmarshal(buff[openP2PHeaderSize:openP2PHeaderSize+int(head.DataLen)], &req); err == nil { tunnelID = req.ID } } else { // compatible with old version tunnelID = t.id } if head.MainType == MsgP2P && head.SubType == MsgPunchHandshakeAck && tunnelID == t.id { break } else { gLog.d("tid:%d handshakeS2C read msg but not MsgPunchHandshakeAck", t.id) } } } if head.MainType == MsgP2P && head.SubType == MsgPunchHandshakeAck { gLog.d("tid:%d handshakeS2C read handshake ack %s", t.id, conn.LocalAddr().String()) UDPWrite(conn, t.remoteHoleAddr, MsgP2P, MsgPunchHandshakeAck, P2PHandshakeReq{ID: t.id}) gotIt = true la, _ := net.ResolveUDPAddr("udp", conn.LocalAddr().String()) gotCh <- la return nil } else { gLog.d("tid:%d handshakeS2C read msg but not MsgPunchHandshakeAck", t.id) } return nil }(t) } gLog.d("tid:%d send symmetric handshake end", t.id) if compareVersion(t.config.peerVersion, SymmetricSimultaneouslySendVersion) < 0 { // compatible with old client gLog.d("tid:%d handshakeS2C ready, notify peer connect", t.id) GNetwork.push(t.config.PeerNode, MsgPushHandshakeStart, TunnelMsg{ID: t.id}) } select { case <-time.After(HandshakeTimeout): return fmt.Errorf("tid:%d wait handshake timeout", t.id) case la := <-gotCh: t.localHoleAddr = la gLog.i("tid:%d handshakeS2C ok. cost %dms", t.id, time.Since(startTime)/time.Millisecond) } return nil }