Modular 、Functional enhancement

client/client.go

@@ -8,7 +8,6 @@ import (
 	"github.com/cnlh/nps/vender/github.com/astaxie/beego/logs"
 	"github.com/cnlh/nps/vender/github.com/xtaci/kcp"
 	"net"
-	"os"
 	"time"
 )
 
@@ -43,106 +42,47 @@ retry:
 		time.Sleep(time.Second * 5)
 		goto retry
 	}
-
 	logs.Info("Successful connection with server %s", s.svrAddr)
+	//monitor the connection
 	go s.ping()
-	s.processor(c)
-}
-
-//处理
-func (s *TRPClient) processor(c *conn.Conn) {
 	s.signal = c
-	go s.dealChan()
+	//start a channel connection
+	go s.newChan()
+	//start health check if the it's open
 	if s.cnf != nil && len(s.cnf.Healths) > 0 {
 		go heathCheck(s.cnf.Healths, s.signal)
 	}
+	//msg connection, eg udp
+	s.handleMain()
+}
+
+//handle main connection
+func (s *TRPClient) handleMain() {
 	for {
-		flags, err := c.ReadFlag()
+		flags, err := s.signal.ReadFlag()
 		if err != nil {
 			logs.Error("Accept server data error %s, end this service", err.Error())
 			break
 		}
 		switch flags {
-		case common.VERIFY_EER:
-			logs.Error("VKey:%s is incorrect, the server refuses to connect, please check", s.vKey)
-			os.Exit(0)
-		case common.RES_CLOSE:
-			logs.Error("The authentication key is connected by another client or the server closes the client.")
-			os.Exit(0)
-		case common.RES_MSG:
-			logs.Error("Server-side return error")
-			break
 		case common.NEW_UDP_CONN:
-			//读取服务端地址、密钥 继续做处理
-			if lAddr, err := c.GetShortLenContent(); err != nil {
+			//read server udp addr and password
+			if lAddr, err := s.signal.GetShortLenContent(); err != nil {
 				logs.Warn(err)
 				return
-			} else if pwd, err := c.GetShortLenContent(); err == nil {
+			} else if pwd, err := s.signal.GetShortLenContent(); err == nil {
 				go s.newUdpConn(string(lAddr), string(pwd))
 			}
-		default:
-			logs.Warn("The error could not be resolved")
-			break
 		}
 	}
-	c.Close()
 	s.Close()
 }
 
 func (s *TRPClient) newUdpConn(rAddr string, md5Password string) {
-	tmpConn, err := common.GetLocalUdpAddr()
-	if err != nil {
-		logs.Error(err)
-		return
-	}
-	localAddr, _ := net.ResolveUDPAddr("udp", tmpConn.LocalAddr().String())
-	localConn, err := net.ListenUDP("udp", localAddr)
-	if err != nil {
-		logs.Error(err)
-		return
-	}
-	localKcpConn, err := kcp.NewConn(rAddr, nil, 150, 3, localConn)
-	if err != nil {
-		logs.Error(err)
-		return
-	}
-	conn.SetUdpSession(localKcpConn)
-	localToolConn := conn.NewConn(localKcpConn)
-	//写入密钥、provider身份
-	if _, err := localToolConn.Write([]byte(md5Password)); err != nil {
-		logs.Error(err)
-		return
-	}
-	if _, err := localToolConn.Write([]byte(common.WORK_P2P_PROVIDER)); err != nil {
-		logs.Error(err)
-		return
-	}
-	//接收服务端传的visitor地址
-	var b []byte
-	if b, err = localToolConn.GetShortLenContent(); err != nil {
-		logs.Error(err)
-		return
-	}
-	//向visitor地址发送测试消息
-	visitorAddr, err := net.ResolveUDPAddr("udp", string(b))
-	if err != nil {
-		logs.Error(err)
-		return
-	}
-	//向目标IP发送探测包
-	if _, err := localConn.WriteTo([]byte("test"), visitorAddr); err != nil {
-		logs.Error(err)
-		return
-	}
-	//给服务端发反馈
-	if _, err := localToolConn.Write([]byte(common.VERIFY_SUCCESS)); err != nil {
-		logs.Error(err)
-		return
-	}
-	//关闭与服务端的连接
-	localConn.Close()
-	//关闭与服务端udp conn，建立新的监听
-	if localConn, err = net.ListenUDP("udp", localAddr); err != nil {
+	var localConn net.PacketConn
+	var err error
+	var remoteAddress string
+	if remoteAddress, localConn, err = handleP2PUdp(rAddr, md5Password, common.WORK_P2P_PROVIDER); err != nil {
 		logs.Error(err)
 		return
 	}
@@ -151,6 +91,7 @@ func (s *TRPClient) newUdpConn(rAddr string, md5Password string) {
 		logs.Error(err)
 		return
 	}
+	logs.Trace("start local p2p udp listen, local address", localConn.LocalAddr().String())
 	//接收新的监听，得到conn，
 	for {
 		udpTunnel, err := l.AcceptKCP()
@@ -159,23 +100,24 @@ func (s *TRPClient) newUdpConn(rAddr string, md5Password string) {
 			l.Close()
 			return
 		}
-		if udpTunnel.RemoteAddr().String() == string(b) {
+		if udpTunnel.RemoteAddr().String() == string(remoteAddress) {
 			conn.SetUdpSession(udpTunnel)
-			//读取link,设置msgCh 设置msgConn消息回传响应机制
+			logs.Trace("successful connection with client ,address %s", udpTunnel.RemoteAddr().String())
+			//read link info from remote
 			l := mux.NewMux(udpTunnel, s.bridgeConnType)
 			for {
 				connMux, err := l.Accept()
 				if err != nil {
 					continue
 				}
-				go s.srcProcess(connMux)
+				go s.handleChan(connMux)
 			}
 		}
 	}
 }
 
 //mux tunnel
-func (s *TRPClient) dealChan() {
+func (s *TRPClient) newChan() {
 	tunnel, err := NewConn(s.bridgeConnType, s.vKey, s.svrAddr, common.WORK_CHAN, s.proxyUrl)
 	if err != nil {
 		logs.Error("connect to ", s.svrAddr, "error:", err)
@@ -189,11 +131,11 @@ func (s *TRPClient) dealChan() {
 			s.Close()
 			break
 		}
-		go s.srcProcess(src)
+		go s.handleChan(src)
 	}
 }
 
-func (s *TRPClient) srcProcess(src net.Conn) {
+func (s *TRPClient) handleChan(src net.Conn) {
 	lk, err := conn.NewConn(src).GetLinkInfo()
 	if err != nil {
 		src.Close()
@@ -218,9 +160,8 @@ loop:
 	for {
 		select {
 		case <-s.ticker.C:
-			if s.tunnel.IsClose {
+			if s.tunnel != nil && s.tunnel.IsClose {
 				s.Close()
-				s.ticker.Stop()
 				break loop
 			}
 		}

client/control.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cnlh/nps/lib/crypt"
 	"github.com/cnlh/nps/lib/version"
 	"github.com/cnlh/nps/vender/github.com/astaxie/beego/logs"
+	"github.com/cnlh/nps/vender/github.com/ccding/go-stun/stun"
 	"github.com/cnlh/nps/vender/github.com/xtaci/kcp"
 	"github.com/cnlh/nps/vender/golang.org/x/net/proxy"
 	"io/ioutil"
@@ -162,7 +163,7 @@ re:
 
 	//create local server secret or p2p
 	for _, v := range cnf.LocalServer {
-		go StartLocalServer(v, cnf.CommonConfig)
+		go startLocalServer(v, cnf.CommonConfig)
 	}
 
 	c.Close()
@@ -238,6 +239,7 @@ func NewConn(tp string, vkey string, server string, connType string, proxyUrl st
 	return c, nil
 }
 
+//http proxy connection
 func NewHttpProxyConn(url *url.URL, remoteAddr string) (net.Conn, error) {
 	req := &http.Request{
 		Method: "CONNECT",
@@ -266,7 +268,143 @@ func NewHttpProxyConn(url *url.URL, remoteAddr string) (net.Conn, error) {
 	return proxyConn, nil
 }
 
+//get a basic auth string
 func basicAuth(username, password string) string {
 	auth := username + ":" + password
 	return base64.StdEncoding.EncodeToString([]byte(auth))
 }
+
+func handleP2PUdp(rAddr, md5Password, role string) (remoteAddress string, c net.PacketConn, err error) {
+	tmpConn, err := common.GetLocalUdpAddr()
+	if err != nil {
+		logs.Error(err)
+		return
+	}
+	localConn, err := newUdpConnByAddr(tmpConn.LocalAddr().String())
+	if err != nil {
+		logs.Error(err)
+		return
+	}
+	localKcpConn, err := kcp.NewConn(rAddr, nil, 150, 3, localConn)
+	if err != nil {
+		logs.Error(err)
+		return
+	}
+	conn.SetUdpSession(localKcpConn)
+	localToolConn := conn.NewConn(localKcpConn)
+	//get local nat type
+	//localNatType, host, err := stun.NewClient().Discover()
+	//if err != nil || host == nil {
+	//	err = errors.New("get nat type error")
+	//	return
+	//}
+	localNatType := stun.NATRestricted
+	//write password
+	if _, err = localToolConn.Write([]byte(md5Password)); err != nil {
+		return
+	}
+	//write role
+	if _, err = localToolConn.Write([]byte(role)); err != nil {
+		return
+	}
+	if err = binary.Write(localToolConn, binary.LittleEndian, int32(localNatType)); err != nil {
+		return
+	}
+	//get another type address and nat type from server
+	var remoteAddr []byte
+	var remoteNatType int32
+	if remoteAddr, err = localToolConn.GetShortLenContent(); err != nil {
+		return
+	}
+	if err = binary.Read(localToolConn, binary.LittleEndian, &remoteNatType); err != nil {
+		return
+	}
+	localConn.Close()
+	//logs.Trace("remote nat type %d,local nat type %s", remoteNatType, localNatType)
+	if remoteAddress, err = sendP2PTestMsg(string(remoteAddr), tmpConn.LocalAddr().String()); err != nil {
+		return
+	}
+	c, err = newUdpConnByAddr(tmpConn.LocalAddr().String())
+	return
+}
+
+func handleP2P(natType1, natType2 int, addr1, addr2 string, role string) (string, error) {
+	switch natType1 {
+	case int(stun.NATFull):
+		return sendP2PTestMsg(addr2, addr1)
+	case int(stun.NATRestricted):
+		switch natType2 {
+		case int(stun.NATFull), int(stun.NATRestricted), int(stun.NATPortRestricted), int(stun.NATSymetric):
+			return sendP2PTestMsg(addr2, addr1)
+		}
+	case int(stun.NATPortRestricted):
+		switch natType2 {
+		case int(stun.NATFull), int(stun.NATRestricted), int(stun.NATPortRestricted):
+			return sendP2PTestMsg(addr2, addr1)
+		}
+	case int(stun.NATSymetric):
+		switch natType2 {
+		case int(stun.NATFull), int(stun.NATRestricted):
+			return sendP2PTestMsg(addr2, addr1)
+		}
+	}
+	return "", errors.New("not support p2p")
+}
+
+func sendP2PTestMsg(remoteAddr string, localAddr string) (string, error) {
+	remoteUdpAddr, err := net.ResolveUDPAddr("udp", remoteAddr)
+	if err != nil {
+		return "", err
+	}
+	localConn, err := newUdpConnByAddr(localAddr)
+	defer localConn.Close()
+	if err != nil {
+		return "", err
+	}
+	buf := make([]byte, 10)
+	for i := 20; i > 0; i-- {
+		logs.Trace("try send test packet to target %s", remoteAddr)
+		if _, err := localConn.WriteTo([]byte(common.WORK_P2P_CONNECT), remoteUdpAddr); err != nil {
+			return "", err
+		}
+		localConn.SetReadDeadline(time.Now().Add(time.Millisecond * 500))
+		n, addr, err := localConn.ReadFromUDP(buf)
+		localConn.SetReadDeadline(time.Time{})
+		switch string(buf[:n]) {
+		case common.WORK_P2P_SUCCESS:
+			for i := 20; i > 0; i-- {
+				if _, err = localConn.WriteTo([]byte(common.WORK_P2P_END), addr); err != nil {
+					return "", err
+				}
+			}
+			return addr.String(), nil
+		case common.WORK_P2P_END:
+			logs.Trace("Remotely Address %s Reply Packet Successfully Received", addr.String())
+			return addr.String(), nil
+		case common.WORK_P2P_CONNECT:
+			go func() {
+				for i := 20; i > 0; i-- {
+					logs.Trace("try send receive success packet to target %s", remoteAddr)
+					if _, err = localConn.WriteTo([]byte(common.WORK_P2P_SUCCESS), addr); err != nil {
+						return
+					}
+					time.Sleep(time.Second)
+				}
+			}()
+		}
+	}
+	localConn.Close()
+	return "", errors.New("connect to the target failed, maybe the nat type is not support p2p")
+}
+
+func newUdpConnByAddr(addr string) (*net.UDPConn, error) {
+	udpAddr, err := net.ResolveUDPAddr("udp", addr)
+	if err != nil {
+		return nil, err
+	}
+	udpConn, err := net.ListenUDP("udp", udpAddr)
+	if err != nil {
+		return nil, err
+	}
+	return udpConn, nil
+}

client/local.go

@@ -11,12 +11,16 @@ import (
 	"github.com/cnlh/nps/vender/github.com/xtaci/kcp"
 	"net"
 	"net/http"
+	"sync"
 )
 
-var LocalServer []*net.TCPListener
-var udpConn net.Conn
-var muxSession *mux.Mux
-var fileServer []*http.Server
+var (
+	LocalServer []*net.TCPListener
+	udpConn     net.Conn
+	muxSession  *mux.Mux
+	fileServer  []*http.Server
+	lock        sync.Mutex
+)
 
 func CloseLocalServer() {
 	for _, v := range LocalServer {
@@ -39,10 +43,10 @@ func startLocalFileServer(config *config.CommonConfig, t *file.Tunnel, vkey stri
 	logs.Info("start local file system, local path %s, strip prefix %s ,remote port %s ", t.LocalPath, t.StripPre, t.Ports)
 	fileServer = append(fileServer, srv)
 	listener := mux.NewMux(remoteConn.Conn, common.CONN_TCP)
-	logs.Warn(srv.Serve(listener))
+	logs.Error(srv.Serve(listener))
 }
 
-func StartLocalServer(l *config.LocalServer, config *config.CommonConfig) error {
+func startLocalServer(l *config.LocalServer, config *config.CommonConfig) error {
 	listener, err := net.ListenTCP("tcp", &net.TCPAddr{net.ParseIP("0.0.0.0"), l.Port, ""})
 	if err != nil {
 		logs.Error("local listener startup failed port %d, error %s", l.Port, err.Error())
@@ -52,15 +56,15 @@ func StartLocalServer(l *config.LocalServer, config *config.CommonConfig) error
 	logs.Info("successful start-up of local monitoring, port", l.Port)
 	conn.Accept(listener, func(c net.Conn) {
 		if l.Type == "secret" {
-			processSecret(c, config, l)
+			handleSecret(c, config, l)
 		} else {
-			processP2P(c, config, l)
+			handleP2PVisitor(c, config, l)
 		}
 	})
 	return nil
 }
 
-func processSecret(localTcpConn net.Conn, config *config.CommonConfig, l *config.LocalServer) {
+func handleSecret(localTcpConn net.Conn, config *config.CommonConfig, l *config.LocalServer) {
 	remoteConn, err := NewConn(config.Tp, config.VKey, config.Server, common.WORK_SECRET, config.ProxyUrl)
 	if err != nil {
 		logs.Error("Local connection server failed ", err.Error())
@@ -73,21 +77,28 @@ func processSecret(localTcpConn net.Conn, config *config.CommonConfig, l *config
 	conn.CopyWaitGroup(remoteConn.Conn, localTcpConn, false, false, nil, nil, false, nil)
 }
 
-func processP2P(localTcpConn net.Conn, config *config.CommonConfig, l *config.LocalServer) {
+func handleP2PVisitor(localTcpConn net.Conn, config *config.CommonConfig, l *config.LocalServer) {
+restart:
+	lock.Lock()
 	if udpConn == nil {
 		newUdpConn(config, l)
 		if udpConn == nil {
+			lock.Unlock()
 			return
 		}
 		muxSession = mux.NewMux(udpConn, "kcp")
 	}
+	lock.Unlock()
+	logs.Trace("start trying to connect with the server")
 	nowConn, err := muxSession.NewConn()
 	if err != nil {
-		logs.Error(err)
+		udpConn = nil
+		logs.Error(err, "reconnect......")
+		goto restart
 		return
 	}
 	//TODO just support compress now because there is not tls file in client packages
-	link := conn.NewLink(common.CONN_TCP, l.Target, false, config.Client.Cnf.Compress, localTcpConn.LocalAddr().String())
+	link := conn.NewLink(common.CONN_TCP, l.Target, false, config.Client.Cnf.Compress, localTcpConn.LocalAddr().String(), false)
 	if _, err := conn.NewConn(nowConn).SendInfo(link, ""); err != nil {
 		logs.Error(err)
 		return
@@ -111,49 +122,18 @@ func newUdpConn(config *config.CommonConfig, l *config.LocalServer) {
 		logs.Error(err)
 		return
 	}
-	//与服务端udp建立连接
-	tmpConn, err := common.GetLocalUdpAddr()
-	if err != nil {
+	var localConn net.PacketConn
+	var remoteAddress string
+	if remoteAddress, localConn, err = handleP2PUdp(string(rAddr), crypt.Md5(l.Password), common.WORK_P2P_VISITOR); err != nil {
+		logs.Error(err)
+		return
+	}
+	udpTunnel, err := kcp.NewConn(remoteAddress, nil, 150, 3, localConn)
+	if err != nil || udpTunnel == nil {
 		logs.Warn(err)
 		return
 	}
-	//与服务端建立udp连接
-	localAddr, _ := net.ResolveUDPAddr("udp", tmpConn.LocalAddr().String())
-	localConn, err := net.ListenUDP("udp", localAddr)
-	if err != nil {
-		logs.Error(err)
-		return
-	}
-	localKcpConn, err := kcp.NewConn(string(rAddr), nil, 150, 3, localConn)
-	if err != nil {
-		logs.Error(err)
-		return
-	}
-	conn.SetUdpSession(localKcpConn)
-	//写入密钥、provider身份
-	if _, err := localKcpConn.Write([]byte(crypt.Md5(l.Password))); err != nil {
-		logs.Error(err)
-		return
-	}
-	if _, err := localKcpConn.Write([]byte(common.WORK_P2P_VISITOR)); err != nil {
-		logs.Error(err)
-		return
-	}
-	//接收服务端传的visitor地址
-	if b, err := conn.NewConn(localKcpConn).GetShortLenContent(); err != nil {
-		logs.Error(err)
-		return
-	} else {
-		//关闭与服务端连接
-		localConn.Close()
-		//建立新的连接
-		localConn, err = net.ListenUDP("udp", localAddr)
-		udpTunnel, err := kcp.NewConn(string(b), nil, 150, 3, localConn)
-		if err != nil || udpTunnel == nil {
-			logs.Warn(err)
-			return
-		}
-		conn.SetUdpSession(udpTunnel)
-		udpConn = udpTunnel
-	}
+	logs.Trace("successful create a connection with server", remoteAddress)
+	conn.SetUdpSession(udpTunnel)
+	udpConn = udpTunnel
 }