3.9.1

USAGE-ZH.md

@@ -96,4 +96,9 @@ firewall-cmd --state
 C:\Program Files\OpenP2P\openp2p.exe uninstall
 # linux,macos
 sudo /usr/local/openp2p/openp2p uninstall
-```
+```
+
+## Docker运行
+```
+docker run -d --net host --name openp2p-client  openp2pcn/openp2p-client:latest -token YOUR-TOKEN -node YOUR-NODE-NAME
+```

USAGE.md

@@ -98,4 +98,9 @@ firewall-cmd --state
 C:\Program Files\OpenP2P\openp2p.exe uninstall
 # linux,macos
 sudo /usr/local/openp2p/openp2p uninstall
+```
+
+## Run with Docker
+```
+docker run -d --net host --name openp2p-client  openp2pcn/openp2p-client:latest -token YOUR-TOKEN -node YOUR-NODE-NAME
 ```

cmd/openp2p.go

@@ -1,7 +1,9 @@
 package main
 
-import openp2p "openp2p/core"
+import (
+	core "openp2p/core"
+)
 
 func main() {
-	openp2p.Run()
+	core.Run()
 }

core/handlepush.go

@@ -43,8 +43,7 @@ func handlePush(pn *P2PNetwork, subType uint16, msg []byte) error {
 		}
 		// verify totp token or token
 		t := totp.TOTP{Step: totp.RelayTOTPStep}
-		if t.Verify(req.Token, pn.config.Token, time.Now().Unix()+(pn.serverTs-pn.localTs)) || // localTs may behind, auto adjust ts
-			t.Verify(req.Token, pn.config.Token, time.Now().Unix()) {
+		if t.Verify(req.Token, pn.config.Token, time.Now().Unix()-pn.dt) { // localTs may behind, auto adjust ts
 			gLog.Printf(LvINFO, "Access Granted\n")
 			config := AppConfig{}
 			config.peerNatType = req.NatType

core/holepunch.go

@@ -23,11 +23,11 @@ func handshakeC2C(t *P2PTunnel) (err error) {
 		gLog.Println(LvDEBUG, "handshakeC2C write MsgPunchHandshake error:", err)
 		return err
 	}
-	ra, head, _, _, err := UDPRead(conn, SymmetricHandshakeAckTimeout)
+	ra, head, _, _, err := UDPRead(conn, HandshakeTimeout)
 	if err != nil {
 		time.Sleep(time.Millisecond * 200)
 		gLog.Println(LvDEBUG, err, ", return this error when ip was not reachable, retry read")
-		ra, head, _, _, err = UDPRead(conn, SymmetricHandshakeAckTimeout)
+		ra, head, _, _, err = UDPRead(conn, HandshakeTimeout)
 		if err != nil {
 			gLog.Println(LvDEBUG, "handshakeC2C read MsgPunchHandshake error:", err)
 			return err
@@ -38,7 +38,7 @@ func handshakeC2C(t *P2PTunnel) (err error) {
 	if head.MainType == MsgP2P && head.SubType == MsgPunchHandshake {
 		gLog.Printf(LvDEBUG, "read %d handshake ", t.id)
 		UDPWrite(conn, t.ra, MsgP2P, MsgPunchHandshakeAck, P2PHandshakeReq{ID: t.id})
-		_, head, _, _, err = UDPRead(conn, SymmetricHandshakeAckTimeout)
+		_, head, _, _, err = UDPRead(conn, HandshakeTimeout)
 		if err != nil {
 			gLog.Println(LvDEBUG, "handshakeC2C write MsgPunchHandshakeAck error", err)
 			return err
@@ -66,7 +66,7 @@ func handshakeC2S(t *P2PTunnel) error {
 	gLog.Printf(LvDEBUG, "handshakeC2S start")
 	defer gLog.Printf(LvDEBUG, "handshakeC2S end")
 	// even if read timeout, continue handshake
-	t.pn.read(t.config.PeerNode, MsgPush, MsgPushHandshakeStart, SymmetricHandshakeAckTimeout)
+	t.pn.read(t.config.PeerNode, MsgPush, MsgPushHandshakeStart, HandshakeTimeout)
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
 	randPorts := r.Perm(65532)
 	conn, err := net.ListenUDP("udp", t.la)
@@ -92,7 +92,7 @@ func handshakeC2S(t *P2PTunnel) error {
 		gLog.Println(LvDEBUG, "send symmetric handshake end")
 		return nil
 	}()
-	deadline := time.Now().Add(SymmetricHandshakeAckTimeout)
+	deadline := time.Now().Add(HandshakeTimeout)
 	err = conn.SetReadDeadline(deadline)
 	if err != nil {
 		gLog.Println(LvERROR, "SymmetricHandshakeAckTimeout SetReadDeadline error")
@@ -140,7 +140,7 @@ func handshakeS2C(t *P2PTunnel) error {
 			}
 			defer conn.Close()
 			UDPWrite(conn, t.ra, MsgP2P, MsgPunchHandshake, P2PHandshakeReq{ID: t.id})
-			_, head, _, _, err := UDPRead(conn, SymmetricHandshakeAckTimeout)
+			_, head, _, _, err := UDPRead(conn, HandshakeTimeout)
 			if err != nil {
 				// gLog.Println(LevelDEBUG, "one of the handshake error:", err)
 				return err
@@ -155,7 +155,7 @@ func handshakeS2C(t *P2PTunnel) error {
 			if head.MainType == MsgP2P && head.SubType == MsgPunchHandshake {
 				gLog.Printf(LvDEBUG, "handshakeS2C read %d handshake ", t.id)
 				UDPWrite(conn, t.ra, MsgP2P, MsgPunchHandshakeAck, P2PHandshakeReq{ID: t.id})
-				_, head, _, _, err = UDPRead(conn, SymmetricHandshakeAckTimeout)
+				_, head, _, _, err = UDPRead(conn, HandshakeTimeout)
 				if err != nil {
 					gLog.Println(LvDEBUG, "handshakeS2C handshake error")
 					return err
@@ -174,7 +174,7 @@ func handshakeS2C(t *P2PTunnel) error {
 	t.pn.push(t.config.PeerNode, MsgPushHandshakeStart, TunnelMsg{ID: t.id})
 
 	select {
-	case <-time.After(SymmetricHandshakeAckTimeout):
+	case <-time.After(HandshakeTimeout):
 		return fmt.Errorf("wait handshake failed")
 	case la := <-gotCh:
 		gLog.Println(LvDEBUG, "symmetric handshake ok", la)

core/log.go

@@ -51,6 +51,7 @@ type logger struct {
 	pid        int
 	maxLogSize int64
 	mode       int
+	stdLogger  *log.Logger
 }
 
 func NewLogger(path string, filePrefix string, level LogLevel, maxLogSize int64, mode int) *logger {
@@ -73,15 +74,16 @@ func NewLogger(path string, filePrefix string, level LogLevel, maxLogSize int64,
 		}
 		os.Chmod(logFilePath, 0644)
 		logfiles[lv] = f
-		loggers[lv] = log.New(f, "", log.LstdFlags)
+		loggers[lv] = log.New(f, "", log.LstdFlags|log.Lmicroseconds)
 	}
 	var le string
 	if runtime.GOOS == "windows" {
 		le = "\r\n"
 	} else {
 		le = "\n"
 	}
-	pLog := &logger{loggers, logfiles, level, logdir, &sync.Mutex{}, le, os.Getpid(), maxLogSize, mode}
+	pLog := &logger{loggers, logfiles, level, logdir, &sync.Mutex{}, le, os.Getpid(), maxLogSize, mode, log.New(os.Stdout, "", 0)}
+	pLog.stdLogger.SetFlags(log.LstdFlags | log.Lmicroseconds)
 	go pLog.checkFile()
 	return pLog
 }
@@ -142,7 +144,7 @@ func (l *logger) Printf(level LogLevel, format string, params ...interface{}) {
 		l.loggers[0].Printf("%d %s "+format+l.lineEnding, params...)
 	}
 	if l.mode == LogConsole || l.mode == LogFileAndConsole {
-		log.Printf("%d %s "+format+l.lineEnding, params...)
+		l.stdLogger.Printf("%d %s "+format+l.lineEnding, params...)
 	}
 }
 
@@ -159,6 +161,6 @@ func (l *logger) Println(level LogLevel, params ...interface{}) {
 		l.loggers[0].Print(params...)
 	}
 	if l.mode == LogConsole || l.mode == LogFileAndConsole {
-		log.Print(params...)
+		l.stdLogger.Print(params...)
 	}
 }

core/nat.go

@@ -3,7 +3,6 @@ package openp2p
 import (
 	"encoding/json"
 	"fmt"
-	"log"
 	"math/rand"
 	"net"
 	"strconv"
@@ -14,20 +13,22 @@ import (
 	reuse "github.com/openp2p-cn/go-reuseport"
 )
 
-func natTCP(serverHost string, serverPort int, localPort int) (publicIP string, publicPort int) {
+func natTCP(serverHost string, serverPort int) (publicIP string, publicPort int, localPort int) {
 	// dialer := &net.Dialer{
 	// 	LocalAddr: &net.TCPAddr{
 	// 		IP:   net.ParseIP("0.0.0.0"),
 	// 		Port: localPort,
 	// 	},
 	// }
-	conn, err := reuse.DialTimeout("tcp4", fmt.Sprintf("%s:%d", "0.0.0.0", localPort), fmt.Sprintf("%s:%d", serverHost, serverPort), time.Second*5)
+	conn, err := reuse.DialTimeout("tcp4", fmt.Sprintf("%s:%d", "0.0.0.0", 0), fmt.Sprintf("%s:%d", serverHost, serverPort), NatTestTimeout)
 	// conn, err := net.Dial("tcp4", fmt.Sprintf("%s:%d", serverHost, serverPort))
+	// log.Println(LvINFO, conn.LocalAddr())
 	if err != nil {
 		fmt.Printf("Dial tcp4 %s:%d error:%s", serverHost, serverPort, err)
 		return
 	}
 	defer conn.Close()
+	localPort, _ = strconv.Atoi(strings.Split(conn.LocalAddr().String(), ":")[1])
 	_, wrerr := conn.Write([]byte("1"))
 	if wrerr != nil {
 		fmt.Printf("Write error: %s\n", wrerr)
@@ -151,14 +152,14 @@ func publicIPTest(publicIP string, echoPort int) (hasPublicIP int, hasUPNPorNATP
 				gLog.Println(LvDEBUG, "could not perform UPNP external address:", err)
 				break
 			}
-			log.Println("PublicIP:", ext)
+			gLog.Println(LvINFO, "PublicIP:", ext)
 
-			externalPort, err := nat.AddPortMapping("udp", echoPort, echoPort, "openp2p", 604800) // 7 days, upnp will perform failed when os start
+			externalPort, err := nat.AddPortMapping("udp", echoPort, echoPort, "openp2p", 30) // 30 seconds fot upnp testing
 			if err != nil {
 				gLog.Println(LvDEBUG, "could not add udp UPNP port mapping", externalPort)
 				break
 			} else {
-				nat.AddPortMapping("tcp", echoPort, echoPort, "openp2p", 604800) // 7 days
+				nat.AddPortMapping("tcp", echoPort, echoPort, "openp2p", 604800) // 7 days for tcp connection
 			}
 		}
 		gLog.Printf(LvDEBUG, "public ip test start %s:%d", publicIP, echoPort)

core/overlay.go

@@ -84,16 +84,14 @@ func (oConn *overlayConn) run() {
 	}
 	oConn.tunnel.overlayConns.Delete(oConn.id)
 	// notify peer disconnect
-	if oConn.isClient {
-		req := OverlayDisconnectReq{ID: oConn.id}
-		if oConn.rtid == 0 {
-			oConn.tunnel.conn.WriteMessage(MsgP2P, MsgOverlayDisconnectReq, &req)
-		} else {
-			// write relay data
-			msg, _ := newMessage(MsgP2P, MsgOverlayDisconnectReq, &req)
-			msgWithHead := append(relayHead.Bytes(), msg...)
-			oConn.tunnel.conn.WriteBytes(MsgP2P, MsgRelayData, msgWithHead)
-		}
+	req := OverlayDisconnectReq{ID: oConn.id}
+	if oConn.rtid == 0 {
+		oConn.tunnel.conn.WriteMessage(MsgP2P, MsgOverlayDisconnectReq, &req)
+	} else {
+		// write relay data
+		msg, _ := newMessage(MsgP2P, MsgOverlayDisconnectReq, &req)
+		msgWithHead := append(relayHead.Bytes(), msg...)
+		oConn.tunnel.conn.WriteBytes(MsgP2P, MsgRelayData, msgWithHead)
 	}
 }
 

core/p2papp.go

@@ -82,7 +82,7 @@ func (app *p2pApp) listenTCP() error {
 			oConn.appKeyBytes = encryptKey
 		}
 		app.tunnel.overlayConns.Store(oConn.id, &oConn)
-		gLog.Printf(LvDEBUG, "Accept TCP overlayID:%d", oConn.id)
+		gLog.Printf(LvDEBUG, "Accept TCP overlayID:%d, %s", oConn.id, oConn.connTCP.RemoteAddr())
 		// tell peer connect
 		req := OverlayConnectReq{ID: oConn.id,
 			Token:    app.tunnel.pn.config.Token,

core/p2pnetwork.go

@@ -25,6 +25,8 @@ var (
 const (
 	retryLimit    = 20
 	retryInterval = 10 * time.Second
+	dtma          = 20
+	ddtma         = 5
 )
 
 type P2PNetwork struct {
@@ -34,9 +36,12 @@ type P2PNetwork struct {
 	restartCh   chan bool
 	wgReconnect sync.WaitGroup
 	writeMtx    sync.Mutex
-	serverTs    int64
-	localTs     int64
 	hbTime      time.Time
+	// for sync server time
+	t1   int64 // nanoSeconds
+	dt   int64 // client faster then server dt nanoSeconds
+	dtma int64
+	ddt  int64 // differential of dt
 	// msgMap    sync.Map
 	msgMap     map[uint64]chan []byte //key: nodeID
 	msgMapMtx  sync.Mutex
@@ -55,6 +60,8 @@ func P2PNetworkInstance(config *NetworkConfig) *P2PNetwork {
 				running:   true,
 				msgMap:    make(map[uint64]chan []byte),
 				limiter:   newBandwidthLimiter(config.ShareBandwidth),
+				dt:        0,
+				ddt:       0,
 			}
 			instance.msgMap[0] = make(chan []byte) // for gateway
 			if config != nil {
@@ -69,11 +76,13 @@ func P2PNetworkInstance(config *NetworkConfig) *P2PNetwork {
 
 func (pn *P2PNetwork) run() {
 	heartbeatTimer := time.NewTicker(NetworkHeartbeatTime)
+	pn.t1 = time.Now().UnixNano()
+	pn.write(MsgHeartbeat, 0, "")
 	for pn.running {
 		select {
 		case <-heartbeatTimer.C:
+			pn.t1 = time.Now().UnixNano()
 			pn.write(MsgHeartbeat, 0, "")
-
 		case <-pn.restartCh:
 			pn.online = false
 			pn.wgReconnect.Wait() // wait read/autorunapp goroutine end
@@ -87,7 +96,7 @@ func (pn *P2PNetwork) run() {
 }
 
 func (pn *P2PNetwork) Connect(timeout int) bool {
-	// waiting for login response
+	// waiting for heartbeat
 	for i := 0; i < (timeout / 1000); i++ {
 		if pn.hbTime.After(time.Now().Add(-NetworkHeartbeatTime)) {
 			return true
@@ -552,8 +561,6 @@ func (pn *P2PNetwork) handleMessage(t int, msg []byte) {
 			gLog.Printf(LvERROR, "login error:%d, detail:%s", rsp.Error, rsp.Detail)
 			pn.running = false
 		} else {
-			pn.serverTs = rsp.Ts
-			pn.hbTime = time.Now()
 			pn.config.Token = rsp.Token
 			pn.config.User = rsp.User
 			gConf.setToken(rsp.Token)
@@ -562,12 +569,28 @@ func (pn *P2PNetwork) handleMessage(t int, msg []byte) {
 				gConf.setNode(rsp.Node)
 				pn.config.Node = rsp.Node
 			}
-			pn.localTs = time.Now().Unix()
-			gLog.Printf(LvINFO, "login ok. user=%s,node=%s,Server ts=%d, local ts=%d", rsp.User, rsp.Node, rsp.Ts, pn.localTs)
+			gLog.Printf(LvINFO, "login ok. user=%s,node=%s", rsp.User, rsp.Node)
 		}
 	case MsgHeartbeat:
 		gLog.Printf(LvDEBUG, "P2PNetwork heartbeat ok")
 		pn.hbTime = time.Now()
+		rtt := pn.hbTime.UnixNano() - pn.t1
+		t2 := int64(binary.LittleEndian.Uint64(msg[openP2PHeaderSize : openP2PHeaderSize+8]))
+		dt := pn.t1 + rtt/2 - t2
+		if pn.dtma == 0 {
+			pn.dtma = dt
+		} else {
+			ddt := dt - pn.dt
+			// if pn.ddt == 0 {
+			pn.ddt = ddt
+			// } else {
+			// 	pn.ddt = pn.ddt/ddtma*(ddtma-1) + ddt/ddtma // avoid int64 overflow
+			// }
+
+			pn.dtma = pn.dtma/dtma*(dtma-1) + dt/dtma // avoid int64 overflow
+		}
+		pn.dt = dt
+		gLog.Printf(LvDEBUG, "server time dt=%dms ddt=%dns rtt=%dms", pn.dt/int64(time.Millisecond), pn.ddt, rtt/int64(time.Millisecond))
 	case MsgPush:
 		handlePush(pn, head.SubType, msg)
 	default: