Skip to content

Commit

Permalink
Modular 、Functional enhancement
Browse files Browse the repository at this point in the history
  • Loading branch information
刘河 committed Apr 8, 2019
1 parent 0c87b41 commit 824b12a
Show file tree
Hide file tree
Showing 41 changed files with 752 additions and 240 deletions.
10 changes: 7 additions & 3 deletions bridge/bridge.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,6 @@ func (s *Bridge) GetHealthFromClient(id int, c *conn.Conn) {
//验证失败，返回错误验证flag，并且关闭连接
func (s *Bridge) verifyError(c *conn.Conn) {
c.Write([]byte(common.VERIFY_EER))
c.Conn.Close()
}

func (s *Bridge) verifySuccess(c *conn.Conn) {
Expand Down Expand Up @@ -291,11 +290,16 @@ func (s *Bridge) register(c *conn.Conn) {
}
}

func (s *Bridge) SendLinkInfo(clientId int, link *conn.Link, linkAddr string, t *file.Tunnel) (target net.Conn, err error) {
func (s *Bridge) SendLinkInfo(clientId int, link *conn.Link, t *file.Tunnel) (target net.Conn, err error) {
//if the proxy type is local
if link.LocalProxy {
target, err = net.Dial(link.ConnType, link.Host)
return
}
if v, ok := s.Client.Load(clientId); ok {
//If ip is restricted to do ip verification
if s.ipVerify {
ip := common.GetIpByAddr(linkAddr)
ip := common.GetIpByAddr(link.RemoteAddr)
if v, ok := s.Register.Load(ip); !ok {
return nil, errors.New(fmt.Sprintf("The ip %s is not in the validation list", ip))
} else {
Expand Down
113 changes: 27 additions & 86 deletions client/client.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)

Expand Down Expand Up @@ -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
}
Expand All @@ -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()
Expand All @@ -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)
Expand All @@ -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()
Expand All @@ -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
}
}
Expand Down
140 changes: 139 additions & 1 deletion client/control.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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
}
Loading

0 comments on commit 824b12a

Please sign in to comment.