diff --git a/.gitignore b/.gitignore index 7d65a58..51e15fd 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,4 @@ src/launchpad.net src/gopkg.in src/ngrok/client/assets/ src/ngrok/server/assets/ +.idea/ \ No newline at end of file diff --git a/src/punching/client/transmit.go b/src/punching/client/frontend.go similarity index 57% rename from src/punching/client/transmit.go rename to src/punching/client/frontend.go index 94405ca..35c270c 100644 --- a/src/punching/client/transmit.go +++ b/src/punching/client/frontend.go @@ -1,17 +1,13 @@ package client import ( - "crypto/md5" - "crypto/rand" - "encoding/base64" - "encoding/hex" "fmt" - "io" "log" "net" - "os" + . "punching/constant" "sync" "time" + "punching/util" ) var ListenAcceptMap map[string]net.Conn @@ -19,34 +15,26 @@ var ExitChanMap map[string]chan bool var RWLock *sync.RWMutex -//生成32位md5字串 -func GetMd5String(s string) string { - h := md5.New() - h.Write([]byte(s)) - return hex.EncodeToString(h.Sum(nil)) -} - -//生成Guid字串 -func UniqueId() string { - b := make([]byte, 48) - - if _, err := io.ReadFull(rand.Reader, b); err != nil { - return "" - } - return GetMd5String(base64.URLEncoding.EncodeToString(b)) -} func handleClientConn(source net.Conn) { - // 32位唯一码 - uniqueid := UniqueId() - log.Println("Enter handleClientConn:", uniqueid) + // 4 bits unique session id + var sessionID string + for{ + RWLock.Lock() + sessionID = util.GenerateRandomPairKey() + if _, ok := ListenAcceptMap[sessionID]; !ok{ + break + } + RWLock.Unlock() + } + log.Println("Enter handleClientConn:", sessionID) RWLock.Lock() - ListenAcceptMap[uniqueid] = source - ExitChanMap[uniqueid] = make(chan bool) + ListenAcceptMap[sessionID] = source + ExitChanMap[sessionID] = make(chan bool) RWLock.Unlock() - log.Println("建立Map", uniqueid) + log.Println("建立Map", sessionID) defer func() { @@ -55,9 +43,9 @@ func handleClientConn(source net.Conn) { log.Println("关闭Sourcer失败") } RWLock.Lock() - delete(ListenAcceptMap, uniqueid) - delete(ExitChanMap, uniqueid) - log.Println("删除map", uniqueid) + delete(ListenAcceptMap, sessionID) + delete(ExitChanMap, sessionID) + log.Println("删除map", sessionID) RWLock.Unlock() }() @@ -75,18 +63,18 @@ func handleClientConn(source net.Conn) { log.Println("读取Source源连接出错，原因为：", err.Error()) //发送控制 - pack01 := Packet(uniqueid, "01", []byte("")) - Wch <- pack01 + packQuit := util.PackageNat(PAIR_CONTROL_QUIT, [4]byte(sessionID),[]byte("") ) + Wch <- packQuit return } - controlID := "00" + controlID := PAIR_CONTROL_NORMAL if flag == 0 { // 第一次 - controlID = "11" + controlID = PAIR_CONTROL_FIRST flag = 1 } - pack := Packet(uniqueid, controlID, buf[0:len01]) + pack := util.PackageNat(controlID, [4]byte(sessionID), buf[0:len01]) Wch <- pack } @@ -94,7 +82,7 @@ func handleClientConn(source net.Conn) { }() select { - case <-ExitChanMap[uniqueid]: + case <-ExitChanMap[sessionID]: log.Println("需要退出Accept") return } @@ -108,6 +96,7 @@ func ClientListenHandle() { RWLock = new(sync.RWMutex) addrOn := Config.Dial + l, err := net.Listen("tcp", addrOn) if err != nil { fmt.Println("listen ", addrOn, " error:", err) @@ -128,45 +117,39 @@ func ClientListenHandle() { go handleClientConn(c) } }() - - select { - case <-Dch: - os.Exit(3) - } - } // 读取目标流到源 func handleReadConn() { for { select { - case r := <-Rch: + case pact := <-Rch: log.Println(time.Now().UnixNano(), "handleReadConn准备处理") // 获取src - controlid := r.ControlID - uniqueid := r.Key - data := r.Data + controlID := pact.ControlID + sessionID := string(pact.SessionID) + data := pact.Data - log.Println("读取Nat包：handleReadConn", uniqueid, "长度为", len(data)) + log.Println("读取Nat包：handleReadConn", sessionID, "长度为", len(data)) //退出 - if controlid == "01" { - if c, ok := ExitChanMap[uniqueid]; ok { + if controlID == PAIR_CONTROL_QUIT { + if c, ok := ExitChanMap[sessionID]; ok { log.Println("发送退出信号") c <- true } else { - log.Println("在ExitChanMap里找不到Key为:", uniqueid) + log.Println("在ExitChanMap里找不到Key为:", sessionID) } } else { - if src, ok := ListenAcceptMap[uniqueid]; ok { + if src, ok := ListenAcceptMap[sessionID]; ok { len2, err2 := src.Write(data) if err2 != nil || len2 <= 0 { log.Println("源写入出错", err2.Error()) } log.Println(time.Now().UnixNano(), "源写入:", len2) } else { - log.Println("在Map里找不到Key为:", uniqueid) + log.Println("在Map里找不到Key为:", sessionID) } } diff --git a/src/punching/client/main.go b/src/punching/client/main.go index 0b03125..4b55931 100644 --- a/src/punching/client/main.go +++ b/src/punching/client/main.go @@ -3,6 +3,15 @@ package client import ( "fmt" "log" + "punching/util" + "time" + "os" +) + +var ( + Dch chan bool + Rch chan util.PairPackage + Wch chan []byte ) func Main() { @@ -13,38 +22,78 @@ func Main() { return } - proxyAddr := Config.Dial || ThirdConfig.Address + proxyAddr := Config.Dial + if proxyAddr == "" { + proxyAddr = ThirdConfig.Address + } pairName := Config.Key - localAddr, destAddr, pairname, err := util.ClientDialProxy(proxyAddr, pairName) + localAddr, destAddr, pairName, err := util.ClientDialProxy(proxyAddr, pairName) if err != nil { log.Println(err) return } - //连接对方 + //连接P2P服务端 connPeer, errPeer := util.DialPeer(localAddr, destAddr) if errPeer != nil { //无法连接上 log.Println(errPeer) return } - // 连接上 P2P客户端 - Dch = make(chan bool) - Rch = make(chan DataPackage) + Rch = make(chan util.PairPackage) Wch = make(chan []byte) - go RHandler(connPeer) - go WHandler(connPeer) + go RHandler(connPeer) //Nat端写通道 + go WHandler(connPeer) //Nat端读通道 + // 侦听端口，开启服务，将端口输入转发到P2P端 ClientListenHandle() + + // 如果P2P端通讯出错，退出 + select { + case <-Dch: + os.Exit(1) + } } func RHandler(conn util.NetConn) { + //声明一个临时缓冲区，用来存储被截断的数据 + tmpBuffer := make([]byte, 0) + + buff := make([]byte, 1024) + for { + j, err := conn.Read(buff) + if err != nil { + log.Println("读取连接数据出错，原因为:", err.Error()) + Dch <- true + break + } + log.Println("准备解包数据:", j) + // 解包 + tmpBuffer = util.UnpackageNat(append(tmpBuffer, buff[:j]...), Rch) + } } func WHandler(conn util.NetConn) { + for { + select { + case msg := <-Wch: + // prefix := string(msg[0:2]) + // if prefix != "00" && prefix != "01" && prefix != "11" { + // log.Println("not equal", string(msg)) + // } else { + l, err := conn.Write(msg) + if err != nil { + log.Println("写到Nat目录连接出错:", err.Error()) + Dch <- true + } else { + log.Println(time.Now().UnixNano(), "已写入到Nat：", l) + } + // } + } + } } diff --git a/src/punching/const/pair.go b/src/punching/const/pair.go deleted file mode 100644 index a5d3e34..0000000 --- a/src/punching/const/pair.go +++ /dev/null @@ -1,19 +0,0 @@ -package const - -// Constant for the client and server - -const ( - PAIR_PACT_HEAD byte = 'P' // C<->S 自定义包头 - PAIR_CONTROL_FIRST byte = 11 // 控制码 C->S第一个包 - PAIR_CONTROL_QUIT byte = 10 // 控制码 退出 - PAIR_CONTROL_NORMAL byte = 0 // 控制码 - - CLIENT_PAIR_ACK byte = 1 // 客户端匹配确认 - SERVER_PAIR_ACK byte = 2 // 服务端匹配确认 -) - -const ( - ROLE_SERVER int = 1 // 点对点服务端 - ROLE_CLIENT int = 2 // 点对点客户端 -} - diff --git a/src/punching/const/proxy.go b/src/punching/const/proxy.go deleted file mode 100644 index 0a84971..0000000 --- a/src/punching/const/proxy.go +++ /dev/null @@ -1,8 +0,0 @@ -package const -const ( - PROXY_PACKAGE_HEAD byte = 'H' // C<->S 自定义包头 - PROXY_CONTROL_ERROR_NO_SERVER byte = 201 // 服务端还没有注册 - PROXY_CONTROL_ERROR_CLIENT_EXIST byte = 202 // 客户端已经存在 - - -) \ No newline at end of file diff --git a/src/punching/constant/pair.go b/src/punching/constant/pair.go new file mode 100644 index 0000000..d35786e --- /dev/null +++ b/src/punching/constant/pair.go @@ -0,0 +1,29 @@ +package constant + +// Constant for the client and server + +const ( + PAIR_CONTROL_FIRST byte = 11 // 控制码 C->S第一个包 + PAIR_CONTROL_QUIT byte = 10 // 控制码 退出 + PAIR_CONTROL_NORMAL byte = 0 // 控制码 + + PAIR_PACKAGE_HEAD_LENGTH = 6 // C<->S 自定义包头长度 + PAIR_PACKAGE_CONTROL_LENGTH = 1 // 包控制码长度 + PAIR_PACKAGE_SESSIONID_LENGTH = 4 // 包会话ID长度 + PAIR_PACKAGE_DATA_LENGTH = 4 // 包数据长度 + PAIR_PACKAGE_PREFIX_LENGTH = 15 // head + control +sessionid + data length + + + +) + +const ( + ROLE_SERVER byte = 1 // 点对点服务端 + ROLE_CLIENT byte = 2 // 点对点客户端 +) + +var ( + PAIR_PACKAGE_HEAD = [6]byte{'C','B','X','N','A','T'} // C<->S 自定义包头 + +) + diff --git a/src/punching/constant/proxy.go b/src/punching/constant/proxy.go new file mode 100644 index 0000000..2cdf16d --- /dev/null +++ b/src/punching/constant/proxy.go @@ -0,0 +1,14 @@ +package constant + +const ( + PROXY_PACKAGE_HEAD byte = 'H' // C<->S 自定义包头 + PROXY_CONTROL_FIRST byte = 11 // 控制ID 第一个数据包 + PROXY_CONTROL_NORMAL byte = 0 // 控制码 正常发送 + PROXY_CONTROL_ACK byte = 12 // 控制码 确认 + PROXY_CONTROL_QUIT byte = 10 // 控制码 退出 + PROXY_CONTROL_ERROR_NO_SERVER byte = 201 // 服务端还没有注册 + PROXY_CONTROL_ERROR_CLIENT_EXIST byte = 202 // 客户端已经存在 + PROXY_CONTROL_ERROR_SERVER_EXIST byte = 203 // 服务端已经存在 + + +) \ No newline at end of file diff --git a/src/punching/proxy/main.go b/src/punching/proxy/main.go index bc04f1d..cc35434 100644 --- a/src/punching/proxy/main.go +++ b/src/punching/proxy/main.go @@ -5,24 +5,11 @@ import ( "sync" "fmt" "punching/util" - "punching/logger" "time" - "punching/const" + . "punching/constant" + "log" ) -const ( - PROXY_FIRST_HEAD byte = 'P' //包头标识 - ROLE_SERVER byte = 1 //服务端 - ROLE_CLIENT byte = 2 //客户端 - PROXY_RESP_NO_SERVER = "no_server_partner" // 代理响应-服务端还没有启用 - PROXY_RESP_CLIENT_EXIST = "client_existed" // 代理响应-客户端已经存在了 - PROXY_RESP_ERR_SERVER_EXIST = "server_existed" // 代理响应-客户端已经存在了 - - CLIENT_RESP_ACK byte = 1 - SERVER_RESP_ACK byte = 1 -) - - // ServerConn 服务端到代理端连接 type ServerConn struct { Rch chan []byte // 读通道 @@ -40,10 +27,10 @@ type ClientConn struct{ // 全局变量 var ( - OnlineServerList map[string]ServerConn // 服务端连接列表Map - OnlineClientList map[string]ClientConn // 客户端连接列表Map - RWLockClient sync.RWMutex //读写锁 - RWLockServer sync.RWMutex + OnlineServerList map[string]*ServerConn // 服务端连接列表Map + OnlineClientList map[string]*ClientConn // 客户端连接列表Map + RWLockClient *sync.RWMutex //读写锁 + RWLockServer *sync.RWMutex ) func Main(){ @@ -54,14 +41,14 @@ func Main(){ return } - - OnlineServerList = make(map[string]ServerConn) - OnlineClientList = make(map[string]ClientConn) + OnlineServerList = make(map[string]*ServerConn) + OnlineClientList = make(map[string]*ClientConn) RWLockClient = new(sync.RWMutex) RWLockServer = new(sync.RWMutex) - tcpAddr, err := net.ResolveTCPAddr("tcp", addr) + listenAddr := Config.Listen + tcpAddr, err := net.ResolveTCPAddr("tcp", listenAddr) if err != nil { panic(err) } @@ -87,19 +74,15 @@ func Main(){ // parseFirstPackage 解析连接的第一个条 // [1]+[1]+[4-32] (包头标识+类型+32字节的Pairname) -func parseFirstPackage(data []byte)(pairname string, roleType int, err error){ - if len(data) <= 6 || len(data) > 34 { - err = fmt.Errorf("%s","包长度不匹配") - return - } +func parseFirstPackage(data []byte)(pairName string, roleType int, err error){ roleType = int(data[1]) - if data[0] != PROXY_FIRST_HEAD || ( roleType != ROLE_CLIENT && roleType != ROLE_SERVER) { + if data[0] != PROXY_PACKAGE_HEAD || ( roleType != ROLE_CLIENT && roleType != ROLE_SERVER) { err = fmt.Errorf("%s","包内容不匹配") return } - pairname = string(byte[2:]) + pairName = string(byte[2:]) return } @@ -184,15 +167,16 @@ func Handler(conn net.Conn) { defer func() { if r := recover(); r != nil { - logger.Println("连接出现问题:%s",r.Error()) + fmt.Printf("连接出现问题:%s",r) } }() defer conn.Close() buf := make([]byte, 1024) - var uid string - var C *OnlineServerSide + + var pairName string + var C *ServerConn // 确定连接类型，判断是否是有效的连接, // 对于客户端，需满足 @@ -211,43 +195,53 @@ func Handler(conn net.Conn) { } //获取匹配名称和连接类型（服务端或客户端) - pairname, roleType, err := parseFirstPackage(buf[0:i]) + pairName, roleType, err := parseFirstPackage(buf[0:i]) if err != nil { return } - + // 处理客户端连接 if roleType == ROLE_CLIENT{ processRoleClient(conn) return // 退出客户端连接 } - - - - - - break - - + break } - // 下面的操作都是针对服务端连接 - // 是否存在pair name - RWLockServer.RLock() - _, ok := OnlineServerList[pairname] - RWLockServer.RUnlock() - // 已存在 - if ok { - conn.Write([]byte(PROXY_RESP_ERR_SERVER_EXIST)) - return + // 服务端连接允许匹配码为空，系统将随机产生唯一匹配码 + if pairName == ""{ + for{ + pairName = util.GenerateRandomPairKey() + RWLockServer.Lock() + if _, ok := OnlineServerList[pairName]; !ok { + break; + } + RWLockServer.Unlock() + } + }else{ + + // 是否存在pair name + RWLockServer.RLock() + _, ok := OnlineServerList[pairName] + RWLockServer.RUnlock() + + // 已存在 + if ok { + errPack := util.PackageProxy(PROXY_CONTROL_ERROR_SERVER_EXIST, []byte("")) + conn.Write(errPack) + fmt.Printf("服务端列表中已存在:%s", pairName) + return + } + } + // 生成服务器连接对象添加到列表 RWLockServer.Lock() - serverConn := &ServerConn{Rch: make(chan []byte), Wch: make(chan []byte), Pairname: pairname, LocalAddr: conn.LocalAddr} - OnlineServerList[pairname] = serverConn + serverConn := &ServerConn{Rch: make(chan []byte), Wch: make(chan []byte), Pairname: pairName, LocalAddr: conn.LocalAddr().String()} + OnlineServerList[pairName] = serverConn RWLockServer.Unlock() // 写通道 diff --git a/src/punching/server/transmit.go b/src/punching/server/backend.go similarity index 99% rename from src/punching/server/transmit.go rename to src/punching/server/backend.go index 192edc1..f0b8838 100644 --- a/src/punching/server/transmit.go +++ b/src/punching/server/backend.go @@ -1,4 +1,4 @@ -package client +package server import ( "log" diff --git a/src/punching/server/main.go b/src/punching/server/main.go index 570bb6e..5f85378 100644 --- a/src/punching/server/main.go +++ b/src/punching/server/main.go @@ -15,9 +15,14 @@ func Main() { return } - proxyAddr := Config.Dial || ThirdConfig.Address + // Proxy Server Address + proxyAddr := Config.Dial + if proxyAddr == "" { + proxyAddr = ThirdConfig.Address + } pairName := Config.Key + // 如果跟Peer连接出错，要重新连接Proxy for { diff --git a/src/punching/server/proxy.go b/src/punching/server/proxy.go index 909ba37..137996d 100644 --- a/src/punching/server/proxy.go +++ b/src/punching/server/proxy.go @@ -4,26 +4,34 @@ import ( "fmt" "log" "net" - "os" + "punching/util" + . "punching/constant" + "strings" +) +var ( + ProxyDch chan byte + ProxyRch chan []byte + ProxyWch chan []byte ) -func WaitForPeer(conn util.NetConn) (localAddr string, remoteAddr string, pairname string, err error) { + +func WaitForPeer(conn util.NetConn) (localAddr string, remoteAddr string, err error) { // 接收心跳和客户端连接确认 - Rch = make(chan DataPackage) - Wch = make(chan []byte) + ProxyRch = make(chan []byte) + ProxyWch = make(chan []byte) + + go RProxyHandler(conn) - go RHandler(conn) - go WHandler(conn) select { - case ret := <-Dch: + case ret := <-ProxyDch: switch ret { - case 10: //无法跟proxy连接 + case PROXY_CONTROL_QUIT: //无法跟proxy连接 //关闭连接 err = error("出现错误") break - case 11: // 获取客户端发来信息 + case PROXY_CONTROL_NORMAL: // 获取客户端发来信息 localAddr = "" remoteAddr = "" pairName = "" @@ -34,13 +42,13 @@ func WaitForPeer(conn util.NetConn) (localAddr string, remoteAddr string, pairna } } -func ServerDialProxy(proxyAddr string, pairkey string) (conn util.NetConn, err error) { +func ServerDialProxy(proxyAddr string, pairName string) (retConn util.NetConn, err error) { // 发第一个包 // 地址接收确认 // 心跑包接收和确认 - var conn = util.NetConn{} + var conn util.NetConn // 不指定端口，让系统自动分配 err = conn.Bind("tcp", "") @@ -57,43 +65,54 @@ func ServerDialProxy(proxyAddr string, pairkey string) (conn util.NetConn, err e fmt.Println("连接服务端出错", err.Error()) return } - defer conn.Close() + fmt.Println("已连接服务器，服务器地址是：%s:%d", tcpAddr.IP.String(), tcpAddr.Port) - // 发送第一个包 - PackageProxy() - key := "ok" + name - buff01 := make([]byte, len(key)) + // 构造自定义包 + data := make([]byte, 4) + data = append(data, []byte{ROLE_SERVER}...) + data = append(data, []byte(pairName)...) - //发送第一个包 - log.Println("发送数据1", name) - n, err := conn.Write([]byte(name)) + packFirst := util.PackageProxy(PROXY_CONTROL_FIRST, data) + + _, err = conn.Write(packFirst) if err != nil { - log.Println("出错，原因：", err.Error()) - os.Exit(22) + return } - log.Println("have write: ", n) - // 读取 - // 错误，退出此次连接 - // 没有错误，显示当前连接的信息 - i, err := conn.Read(buff01) + buff := make([]byte, 1024) + + // 获取返回信息 + i, err := conn.Read(buff) if err != nil { fmt.Println("读取数据出错，", err.Error()) - os.Exit(1) + return } - // log.Print("读取数据:", string(buff01[0:i])) - // if string(buff01[0:i]) == key { - // break - // } - // log.Println("发送数据", []byte(name)) - // conn.Write([]byte(name)) + controlID := buff[1] + switch controlID { + case PROXY_CONTROL_NORMAL: + retData := string(buff[2:i]) + items := strings.Split(retData, ",") + localAddr := items[0] + rePairName := items[1] + + fmt.Printf("P2P服务端侦听地址为：%s, 匹配码为:%s", localAddr, rePairName) + + break + case PROXY_CONTROL_ERROR_SERVER_EXIST: + err = fmt.Errorf("错误，P2P服务端已存在") + break + default: + err = fmt.Errorf("无效的控制码,%d",int(controlID)) + } + + return conn, nil } -func RHandler(conn util.NetConn) { +func RProxyHandler(conn util.NetConn) { for { // 心跳包,回复ack diff --git a/src/punching/util/net.go b/src/punching/util/net.go index 64cda29..d0a6c6a 100644 --- a/src/punching/util/net.go +++ b/src/punching/util/net.go @@ -4,34 +4,32 @@ import ( "fmt" "log" "net" - "os" - "p2p/util" - "strings" "time" + . "punching/constant" + "strings" ) // 连接到Proxy代理解析端， 地址/类型/IP // 发送第一个包->(接收到错误数据，退出，否则接收到NatAddress,->确认收到 <->完成 不断接收心跳包，确认) // 客户端 错误失败 | nat地址 // 服务端 错误失败 | 注册 心跑包 -func ClientDialProxy(proxyAddr string, pairkey string, role string) (localAddr string, destAddr string, pairname string, err error) { - // 发第一个包 - // 地址接收确认 - // 心跑包接收和确认 +// ClientDialProxy P2P客户端连接到Proxy +// 连接成功后获取本地地址，远程地址和匹配码，否则将返回错误 +func ClientDialProxy(proxyAddr string, pairName string) (localAddr string, remoteAddr string, rePairName string, err error) { - var conn = util.NetConn{} + var conn = NetConn{} // 不指定端口，让系统自动分配 err = conn.Bind("tcp", "") if err != nil { - log.Println("绑定出错", err.Error()) + fmt.Println("绑定出错", err.Error()) return } // 连接到Proxy解析服务器 tcpAddr, err := net.ResolveTCPAddr("tcp", proxyAddr) - err = conn.Connect(util.InetAddr(tcpAddr.IP.String()), tcpAddr.Port) + err = conn.Connect(InetAddr(tcpAddr.IP.String()), tcpAddr.Port) if err != nil { fmt.Println("连接服务端出错", err.Error()) @@ -40,69 +38,62 @@ func ClientDialProxy(proxyAddr string, pairkey string, role string) (localAddr s defer conn.Close() fmt.Println("已连接服务器，服务器地址是：%s:%d", tcpAddr.IP.String(), tcpAddr.Port) - for { + // 构造自定义包 + data := make([]byte, 4) + data = append(data, []byte{ROLE_CLIENT}...) + data = append(data, []byte(pairName)...) - PackageProxy() + packFirst := PackageProxy(PROXY_CONTROL_FIRST, data) - key := "ok" + name - buff01 := make([]byte, len(key)) - - //发送第一个包 - log.Println("发送数据1", name) - n, err := conn.Write([]byte(name)) - if err != nil { - log.Println("出错，原因：", err.Error()) - os.Exit(22) - } - log.Println("have write: ", n) - - i, err := conn.Read(buff01) - if err != nil { - fmt.Println("读取数据出错，", err.Error()) - os.Exit(1) - } - log.Print("读取数据:", string(buff01[0:i])) - if string(buff01[0:i]) == key { - break - } - log.Println("发送数据", []byte(name)) - conn.Write([]byte(name)) - } - - fmt.Println("--- Request sent, waiting for parkner in Name ...", name) - - buff := make([]byte, 512) - - //获取服务发送过来的对方IP信息 - len, err := conn.Read(buff) + _, err = conn.Write(packFirst) if err != nil { - log.Println("客户端读取数据出错:", err.Error()) - os.Exit(1) + return } - pairAddr := string(buff[0:len]) - arrPairAddr := strings.Split(pairAddr, ",") - remoteAddrStr := arrPairAddr[0] - localAddrStr := arrPairAddr[1] - log.Println("读取到远程IP", remoteAddrStr) + buff := make([]byte, 1024) - conn.Close() - //time.Sleep(3 * time.Second) + // 获取返回信息 + i, err := conn.Read(buff) + if err != nil { + fmt.Println("读取数据出错，", err.Error()) + return + } - remoteAddr, err := net.ResolveTCPAddr("tcp", remoteAddrStr) + controlID := buff[1] + switch controlID { + case PROXY_CONTROL_NORMAL: + retData := string(buff[2:i]) + items := strings.Split(retData, ",") + localAddr = items[0] + remoteAddr = items[1] + rePairName = items[2] - //localAddr.String() - log.Println("local addr:", localAddrStr) + // 发送确认 + packAck := PackageProxy(PROXY_CONTROL_ACK, []byte("")) + conn.Write(packAck) + break + case PROXY_CONTROL_ERROR_NO_SERVER: + err = fmt.Errorf("错误，P2P服务端不存在") + break + case PROXY_CONTROL_ERROR_CLIENT_EXIST: + err = fmt.Errorf("错误，P2P服务端不存在") + break + default: + err = fmt.Errorf("无效的控制码,%d",int(controlID)) + } + return } - - -func DialPeer(localAddr string, remoteAddr string) util.NetConn { +func DialPeer(localAddr string, remoteAddr string) (netconn NetConn, err error) { remoteTCPAddr, err := net.ResolveTCPAddr("tcp", remoteAddr) + if err != nil { + log.Println("The format of remote address is invalid, %s", err.Error()) + return + } - var conn util.NetConn + var conn NetConn // 不指定端口，让系统自动分配 err = conn.Bind("tcp", localAddr) @@ -111,19 +102,19 @@ func DialPeer(localAddr string, remoteAddr string) util.NetConn { return } - // TODO 应该封装成一个函数 - // 有时连接并不成功，多次连接 - log.Println("远程地址是:", remoteAddr.IP.String(), remoteAddr.Port) + log.Println("远程地址是:", remoteTCPAddr.IP.String(), remoteTCPAddr.Port) + // 有时连接一次并不成功，尝试多次连接 tryCount := 0 for { tryCount += 1 if tryCount > 10 { - log.Printf("连接不上，退出") + err = fmt.Errorf("Attempt to connect remote address, but failed, local addrss: %s, "+ + "remote address", localAddr, remoteAddr) return } - err02 := conn02.Connect(util.InetAddr(remoteAddr.IP.String()), remoteAddr.Port) + err02 := conn.Connect(util.InetAddr(remoteTCPAddr.IP.String()), remoteTCPAddr.Port) if err02 != nil { log.Printf("第%d次不能连接远程服务器:%s", tryCount, err02.Error()) time.Sleep(1 * time.Second) @@ -133,5 +124,5 @@ func DialPeer(localAddr string, remoteAddr string) util.NetConn { break } } - + return conn, nil } diff --git a/src/punching/util/package.go b/src/punching/util/package.go index 3120f1a..be01ce2 100644 --- a/src/punching/util/package.go +++ b/src/punching/util/package.go @@ -1,61 +1,87 @@ -package util +package util -const { - PACKAGE_PROXY_HEADER byte = 'P' - PACKAGE_NAT_HEADER [6]byte = []byte{'C','B','X','N','A','T'} +import ( + "bytes" + . "punching/constant" + "log" +) + +// PairPackage P2P端通讯封装包 +type PairPackage struct { + Head [6]byte // 头 + ControlID byte // 控制ID + SessionID [4]byte // 会话ID + Data []byte // 数据 } // 跟代理解析端通讯封包 -func PackageProxy(control byte, data []byte) []byte{ +func PackageProxy(control byte, data []byte) []byte { pack := bytes.NewBuffer(nil) - pack.Write(PACKAGE_PROXY_HEADER) - pack.Write(control) + pack.Write([]byte{PROXY_PACKAGE_HEAD}) + pack.Write([]byte{control}) pack.Write(data) return pack.Bytes() } -// 跟代理解析端通讯拆包 -func UnpackageProxy(pact []byte)(control byte, data []byte) { - +// 跟代理解析端拆包 +func UnpackageProxy(pact []byte) (control byte, data []byte) { + return pact[1], pact[1:] } -// Nat网络后面的Client端和Server端封包 -func PackageNat(control byte, sessionKey [4]byte, data []byte) []byte{ - +// Customize P2P data package +// The format of package defined below: +// head(6)+control(1)+session id(4) + data length (4) + data +func PackageNat(control byte, sessionID [4]byte, data []byte) []byte { + pack := bytes.NewBuffer(nil) + pack.Write([]byte(PAIR_PACKAGE_HEAD)) // Head [6]byte + pack.Write([]byte{control}) + pack.Write([]byte(sessionID)) + pack.Write(IntToBytes(len(data))) // length of sent data + pack.Write([]byte(sessionID)) + + pack.Write(data) + return pack.Bytes() } // Nat网络后面的Client端和Server端拆包 // 需要考虑沾包，分析出的完整封装包传入读channel -func UnpackageNat(pact []byte)(data []byte){ +func UnpackageNat(buffer []byte, readChan chan PairPackage) (data []byte) { length := len(buffer) log.Println("长度为:", length) var i int for i = 0; i < length; i = i + 1 { - if length < i+ConstHeaderLength+ConstControlLength+ConstDataLength { + if length < i + PAIR_PACKAGE_PREFIX_LENGTH { break } - if string(buffer[i:i+len(ConstHeaderKey)]) == ConstHeaderKey { - // 获取头数据长度 - messageLength := BytesToInt(buffer[i+ConstHeaderLength+ConstControlLength : i+ConstHeaderLength+ConstControlLength+ConstDataLength]) - - log.Println("数据长度为:", messageLength) - log.Println("需要的包长度:", ConstHeaderLength+ConstControlLength+ConstDataLength+messageLength) - if length < i+ConstHeaderLength+ConstControlLength+ConstDataLength+messageLength { + + if string(buffer[i:i+ PAIR_PACKAGE_HEAD_LENGTH]) == string(PAIR_PACKAGE_HEAD) { + // Length of data + dataLength := BytesToInt(buffer[i+ PAIR_PACKAGE_PREFIX_LENGTH - PAIR_PACKAGE_DATA_LENGTH: i + + PAIR_PACKAGE_PREFIX_LENGTH]) + + if length < i+ PAIR_PACKAGE_PREFIX_LENGTH + dataLength { break } - data := buffer[i+ConstHeaderLength+ConstControlLength+ConstDataLength : i+ConstHeaderLength+ConstControlLength+ConstDataLength+messageLength] - controlID := string(buffer[i+ConstHeaderLength : i+ConstHeaderLength+ConstControlLength]) - key := string(buffer[i+len(ConstHeaderKey) : i+ConstHeaderLength]) - log.Println("控制ID为：", controlID, "key为", key) - dataPackage := DataPackage{ - Data: data, - ControlID: controlID, - Key: key, - } - Rch <- dataPackage - i += ConstHeaderLength + ConstControlLength + ConstDataLength + messageLength - 1 + // data + data := buffer[i + PAIR_PACKAGE_PREFIX_LENGTH: i + PAIR_PACKAGE_PREFIX_LENGTH + + dataLength] + + controlID := buffer[i + PAIR_PACKAGE_HEAD_LENGTH : i + PAIR_PACKAGE_HEAD_LENGTH + + PAIR_PACKAGE_CONTROL_LENGTH] + sessionID := string(buffer[i+ PAIR_PACKAGE_HEAD_LENGTH + PAIR_PACKAGE_CONTROL_LENGTH: + i + PAIR_PACKAGE_HEAD_LENGTH + PAIR_PACKAGE_CONTROL_LENGTH + PAIR_PACKAGE_SESSIONID_LENGTH]) + + pact := PairPackage{ + Head: PAIR_PACKAGE_HEAD, + Data: data, + ControlID: controlID[0], + SessionID: [4]byte(sessionID), + } + readChan <- pact + + i += PAIR_PACKAGE_PREFIX_LENGTH + dataLength - 1 } } @@ -64,4 +90,4 @@ func UnpackageNat(pact []byte)(data []byte){ } return buffer[i:] -} \ No newline at end of file +} diff --git a/src/punching/util/rand.go b/src/punching/util/rand.go index 2b56348..829359d 100644 --- a/src/punching/util/rand.go +++ b/src/punching/util/rand.go @@ -1,30 +1,10 @@ package util import ( - "crypto/md5" - "crypto/rand" - "encoding/base64" - "encoding/hex" - "io" + "math/rand" "time" ) -//生成32位md5字串 -func GetMd5String(s string) string { - h := md5.New() - h.Write([]byte(s)) - return hex.EncodeToString(h.Sum(nil)) -} - -//生成Guid字串 -func UniqueId() string { - b := make([]byte, 48) - - if _, err := io.ReadFull(rand.Reader, b); err != nil { - return "" - } - return GetMd5String(base64.URLEncoding.EncodeToString(b)) -} // GenerateRandomPairKey 获取4位随机匹配码 func GenerateRandomPairKey() string { @@ -32,20 +12,21 @@ func GenerateRandomPairKey() string { rndNums := GenerateRandomNumber(97, 122, 4) key := "" for num, _ := range rndNums { - key = key + string(byte{num}) + key = key + string(byte(num)) } return key } //生成count个[start,end)结束的不重复的随机数 func GenerateRandomNumber(start int, end int, count int) []int { - //范围检查 + // Check the range if end < start || (end-start) < count { return nil } - //存放结果的slice + // Slice to store the result nums := make([]int, 0) //随机数生成器，加入时间戳保证每次生成的随机数不一样 + r := rand.New(rand.NewSource(time.Now().UnixNano())) for len(nums) < count { //生成随机数