Merge remote-tracking branch 'origin/dev' into dev

# Conflicts: # server/handler/pool_test.go
2025-08-08 11:10:14 +08:00 · 2021-08-01 21:00:07 +08:00
parent 0f783cfaf6 afe447ada7
commit 2a66df55b0
24 changed files with 14357 additions and 285 deletions
--- a/server/handler/link_cstp.go
+++ b/server/handler/link_cstp.go
@@ -34,9 +34,8 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
 			return
 		}
 		// hdata := make([]byte, BufferSize)
-		hb := getByteFull()
-		hdata := *hb
-		n, err = conn.Read(hdata)
+		pl := getPayload()
+		n, err = conn.Read(pl.Data)
 		if err != nil {
 			base.Error("read hdata: ", err)
 			return
@@ -48,7 +47,7 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
 			base.Error(err)
 		}

-		switch hdata[6] {
+		switch pl.Data[6] {
 		case 0x07: // KEEPALIVE
 			// do nothing
 			// base.Debug("recv keepalive", cSess.IpAddr)
@@ -57,19 +56,24 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
 			return
 		case 0x03: // DPD-REQ
 			// base.Debug("recv DPD-REQ", cSess.IpAddr)
-			if payloadOutCstp(cSess, sessdata.LTypeIPData, 0x04, nil) {
+			pl.PType = 0x04
+			if payloadOutCstp(cSess, pl) {
 				return
 			}
 		case 0x04:
 			// log.Println("recv DPD-RESP")
 		case 0x00: // DATA
-			dataLen = binary.BigEndian.Uint16(hdata[4:6]) // 4,5
-			if payloadIn(cSess, sessdata.LTypeIPData, 0x00, hdata[8:8+dataLen]) {
+			// 获取数据长度
+			dataLen = binary.BigEndian.Uint16(pl.Data[4:6]) // 4,5
+			// 去除数据头
+			copy(pl.Data, pl.Data[8:8+dataLen])
+			// 更新切片长度
+			pl.Data = pl.Data[:dataLen]
+			// pl.Data = append(pl.Data[:0], pl.Data[8:8+dataLen]...)
+			if payloadIn(cSess, pl) {
 				return
 			}
 		}
-
-		putByte(hb)
 	}
 }

@@ -83,37 +87,44 @@ func cstpWrite(conn net.Conn, cSess *sessdata.ConnSession) {
 	var (
 		err error
 		n   int
-		// header  []byte
-		payload *sessdata.Payload
+		pl  *sessdata.Payload
 	)

 	for {
 		select {
-		case payload = <-cSess.PayloadOutCstp:
+		case pl = <-cSess.PayloadOutCstp:
 		case <-cSess.CloseChan:
 			return
 		}

-		if payload.LType != sessdata.LTypeIPData {
+		if pl.LType != sessdata.LTypeIPData {
 			continue
 		}

-		h := []byte{'S', 'T', 'F', 0x01, 0x00, 0x00, payload.PType, 0x00}
-		hb := getByteZero()
-		header := *hb
-		header = append(header, h...)
-		if payload.PType == 0x00 { // data
-			binary.BigEndian.PutUint16(header[4:6], uint16(len(payload.Data)))
-			header = append(header, payload.Data...)
+		if pl.PType == 0x00 {
+			// 获取数据长度
+			l := len(pl.Data)
+			// 先扩容 +8
+			pl.Data = pl.Data[:l+8]
+			// 数据后移
+			copy(pl.Data[8:], pl.Data)
+			// 添加头信息
+			copy(pl.Data[:8], plHeader)
+			// 更新头长度
+			binary.BigEndian.PutUint16(pl.Data[4:6], uint16(l))
+		} else {
+			pl.Data = append(pl.Data[:0], plHeader...)
+			// 设置头类型
+			pl.Data[6] = pl.PType
 		}
-		n, err = conn.Write(header)
+
+		n, err = conn.Write(pl.Data)
 		if err != nil {
 			base.Error("write err", err)
 			return
 		}

-		putByte(hb)
-		putPayload(payload)
+		putPayload(pl)

 		// 限流设置
 		err = cSess.RateLimit(n, false)
--- a/server/handler/link_dtls.go
+++ b/server/handler/link_dtls.go
@@ -24,21 +24,22 @@ func LinkDtls(conn net.Conn, cSess *sessdata.ConnSession) {
 	}()

 	var (
+		err  error
+		n    int
 		dead = time.Duration(cSess.CstpDpd+5) * time.Second
 	)

 	go dtlsWrite(conn, dSess, cSess)

 	for {
-		err := conn.SetReadDeadline(time.Now().Add(dead))
+		err = conn.SetReadDeadline(time.Now().Add(dead))
 		if err != nil {
 			base.Error("SetDeadline: ", err)
 			return
 		}

-		hb := getByteFull()
-		hdata := *hb
-		n, err := conn.Read(hdata)
+		pl := getPayload()
+		n, err = conn.Read(pl.Data)
 		if err != nil {
 			base.Error("read hdata: ", err)
 			return
@@ -50,7 +51,7 @@ func LinkDtls(conn net.Conn, cSess *sessdata.ConnSession) {
 			base.Error(err)
 		}

-		switch hdata[0] {
+		switch pl.Data[0] {
 		case 0x07: // KEEPALIVE
 			// do nothing
 			// base.Debug("recv keepalive", cSess.IpAddr)
@@ -59,18 +60,23 @@ func LinkDtls(conn net.Conn, cSess *sessdata.ConnSession) {
 			return
 		case 0x03: // DPD-REQ
 			// base.Debug("recv DPD-REQ", cSess.IpAddr)
-			if payloadOutDtls(cSess, dSess, sessdata.LTypeIPData, 0x04, nil) {
+			pl.PType = 0x04
+			if payloadOutDtls(cSess, dSess, pl) {
 				return
 			}
 		case 0x04:
 			// base.Debug("recv DPD-RESP", cSess.IpAddr)
 		case 0x00: // DATA
-			if payloadIn(cSess, sessdata.LTypeIPData, 0x00, hdata[1:n]) {
+			// 去除数据头
+			// copy(pl.Data, pl.Data[1:n])
+			// 更新切片长度
+			// pl.Data = pl.Data[:n-1]
+			pl.Data = append(pl.Data[:0], pl.Data[1:n]...)
+			if payloadIn(cSess, pl) {
 				return
 			}
 		}

-		putByte(hb)
 	}
 }

@@ -82,37 +88,42 @@ func dtlsWrite(conn net.Conn, dSess *sessdata.DtlsSession, cSess *sessdata.ConnS
 	}()

 	var (
-		// header  []byte
-		payload *sessdata.Payload
+		pl *sessdata.Payload
 	)

 	for {
 		// dtls优先推送数据
 		select {
-		case payload = <-cSess.PayloadOutDtls:
+		case pl = <-cSess.PayloadOutDtls:
 		case <-dSess.CloseChan:
 			return
 		}

-		if payload.LType != sessdata.LTypeIPData {
+		if pl.LType != sessdata.LTypeIPData {
 			continue
 		}

 		// header = []byte{payload.PType}
-		hb := getByteZero()
-		header := *hb
-		header = append(header, payload.PType)
-		if payload.PType == 0x00 { // data
-			header = append(header, payload.Data...)
+		if pl.PType == 0x00 { // data
+			// 获取数据长度
+			l := len(pl.Data)
+			// 先扩容 +1
+			pl.Data = pl.Data[:l+1]
+			// 数据后移
+			copy(pl.Data[1:], pl.Data)
+			// 添加头信息
+			pl.Data[0] = pl.PType
+		} else {
+			// 设置头类型
+			pl.Data = append(pl.Data[:0], pl.PType)
 		}
-		n, err := conn.Write(header)
+		n, err := conn.Write(pl.Data)
 		if err != nil {
 			base.Error("write err", err)
 			return
 		}

-		putByte(hb)
-		putPayload(payload)
+		putPayload(pl)

 		// 限流设置
 		err = cSess.RateLimit(n, false)
--- a/server/handler/link_tap.go
+++ b/server/handler/link_tap.go
@@ -88,14 +88,14 @@ func tapWrite(ifce *water.Interface, cSess *sessdata.ConnSession) {
 	}()

 	var (
-		err     error
-		payload *sessdata.Payload
-		frame   ethernet.Frame
+		err   error
+		pl    *sessdata.Payload
+		frame ethernet.Frame
 	)

 	for {
 		select {
-		case payload = <-cSess.PayloadIn:
+		case pl = <-cSess.PayloadIn:
 		case <-cSess.CloseChan:
 			return
 		}
@@ -103,17 +103,15 @@ func tapWrite(ifce *water.Interface, cSess *sessdata.ConnSession) {
 		// var frame ethernet.Frame
 		fb := getByteFull()
 		frame = *fb
-		switch payload.LType {
+		switch pl.LType {
 		default:
 			// log.Println(payload)
 		case sessdata.LTypeEthernet:
-			copy(frame, payload.Data)
-			frame = frame[:len(payload.Data)]
+			copy(frame, pl.Data)
+			frame = frame[:len(pl.Data)]
 		case sessdata.LTypeIPData: // 需要转换成 Ethernet 数据
-			data := payload.Data
-
-			ip_src := waterutil.IPv4Source(data)
-			if waterutil.IsIPv6(data) || !ip_src.Equal(cSess.IpAddr) {
+			ip_src := waterutil.IPv4Source(pl.Data)
+			if waterutil.IsIPv6(pl.Data) || !ip_src.Equal(cSess.IpAddr) {
 				// 过滤掉IPv6的数据
 				// 非分配给客户端ip，直接丢弃
 				continue
@@ -122,7 +120,7 @@ func tapWrite(ifce *water.Interface, cSess *sessdata.ConnSession) {
 			// packet := gopacket.NewPacket(data, layers.LayerTypeIPv4, gopacket.Default)
 			// fmt.Println("get:", packet)

-			ip_dst := waterutil.IPv4Destination(data)
+			ip_dst := waterutil.IPv4Destination(pl.Data)
 			// fmt.Println("get:", ip_src, ip_dst)

 			var dstHw net.HardwareAddr
@@ -142,8 +140,8 @@ func tapWrite(ifce *water.Interface, cSess *sessdata.ConnSession) {
 			}
 			// fmt.Println("Gateway", ip_dst, dstAddr.HardwareAddr)

-			frame.Prepare(dstHw, cSess.MacHw, ethernet.NotTagged, ethernet.IPv4, len(data))
-			copy(frame[12+2:], data)
+			frame.Prepare(dstHw, cSess.MacHw, ethernet.NotTagged, ethernet.IPv4, len(pl.Data))
+			copy(frame[12+2:], pl.Data)
 		}

 		// packet := gopacket.NewPacket(frame, layers.LayerTypeEthernet, gopacket.Default)
@@ -155,7 +153,7 @@ func tapWrite(ifce *water.Interface, cSess *sessdata.ConnSession) {
 		}

 		putByte(fb)
-		putPayload(payload)
+		putPayload(pl)
 	}
 }

@@ -168,7 +166,7 @@ func tapRead(ifce *water.Interface, cSess *sessdata.ConnSession) {
 	var (
 		err   error
 		n     int
-		buf   []byte
+		data  []byte
 		frame ethernet.Frame
 	)

@@ -193,7 +191,7 @@ func tapRead(ifce *water.Interface, cSess *sessdata.ConnSession) {
 			continue
 		case ethernet.IPv4:
 			// 发送IP数据
-			data := frame.Payload()
+			data = frame.Payload()

 			ip_dst := waterutil.IPv4Destination(data)
 			if !ip_dst.Equal(cSess.IpAddr) {
@@ -205,7 +203,12 @@ func tapRead(ifce *water.Interface, cSess *sessdata.ConnSession) {
 			// packet := gopacket.NewPacket(data, layers.LayerTypeIPv4, gopacket.Default)
 			// fmt.Println("put:", packet)

-			if payloadOut(cSess, sessdata.LTypeIPData, 0x00, data) {
+			pl := getPayload()
+			// 拷贝数据到pl
+			copy(pl.Data, data)
+			// 更新切片长度
+			pl.Data = pl.Data[:len(data)]
+			if payloadOut(cSess, pl) {
 				return
 			}

@@ -226,7 +229,7 @@ func tapRead(ifce *water.Interface, cSess *sessdata.ConnSession) {
 			// 返回ARP数据
 			src := &arpdis.Addr{IP: cSess.IpAddr, HardwareAddr: cSess.MacHw}
 			dst := &arpdis.Addr{IP: arpReq.SourceProtAddress, HardwareAddr: frame.Source()}
-			buf, err = arpdis.NewARPReply(src, dst)
+			data, err = arpdis.NewARPReply(src, dst)
 			if err != nil {
 				base.Error(err)
 				return
@@ -243,7 +246,15 @@ func tapRead(ifce *water.Interface, cSess *sessdata.ConnSession) {
 			copy(addr.HardwareAddr, frame.Source())
 			arpdis.Add(addr)

-			if payloadIn(cSess, sessdata.LTypeEthernet, 0x00, buf) {
+			pl := getPayload()
+			// 设置为二层数据类型
+			pl.LType = sessdata.LTypeEthernet
+			// 拷贝数据到pl
+			copy(pl.Data, data)
+			// 更新切片长度
+			pl.Data = pl.Data[:len(data)]
+
+			if payloadIn(cSess, pl) {
 				return
 			}

--- a/server/handler/link_tun.go
+++ b/server/handler/link_tun.go
@@ -69,24 +69,24 @@ func tunWrite(ifce *water.Interface, cSess *sessdata.ConnSession) {
 	}()

 	var (
-		err     error
-		payload *sessdata.Payload
+		err error
+		pl  *sessdata.Payload
 	)

 	for {
 		select {
-		case payload = <-cSess.PayloadIn:
+		case pl = <-cSess.PayloadIn:
 		case <-cSess.CloseChan:
 			return
 		}

-		_, err = ifce.Write(payload.Data)
+		_, err = ifce.Write(pl.Data)
 		if err != nil {
 			base.Error("tun Write err", err)
 			return
 		}

-		putPayload(payload)
+		putPayload(pl)
 	}
 }

@@ -102,14 +102,16 @@ func tunRead(ifce *water.Interface, cSess *sessdata.ConnSession) {

 	for {
 		// data := make([]byte, BufferSize)
-		hb := getByteFull()
-		data := *hb
-		n, err = ifce.Read(data)
+		pl := getPayload()
+		n, err = ifce.Read(pl.Data)
 		if err != nil {
 			base.Error("tun Read err", n, err)
 			return
 		}

+		// 更新数据长度
+		pl.Data = (pl.Data)[:n]
+
 		// data = data[:n]
 		// ip_src := waterutil.IPv4Source(data)
 		// ip_dst := waterutil.IPv4Destination(data)
@@ -118,10 +120,8 @@ func tunRead(ifce *water.Interface, cSess *sessdata.ConnSession) {
 		// packet := gopacket.NewPacket(data, layers.LayerTypeIPv4, gopacket.Default)
 		// fmt.Println("read:", packet)

-		if payloadOut(cSess, sessdata.LTypeIPData, 0x00, data[:n]) {
+		if payloadOut(cSess, pl) {
 			return
 		}
-
-		putByte(hb)
 	}
 }
--- a/server/handler/payload.go
+++ b/server/handler/payload.go
@@ -6,18 +6,9 @@ import (
 	"github.com/songgao/water/waterutil"
 )

-func payloadIn(cSess *sessdata.ConnSession, lType sessdata.LType, pType byte, data []byte) bool {
-	pl := getPayload()
-	pl.LType = lType
-	pl.PType = pType
-	pl.Data = append(pl.Data, data...)
-
-	return payloadInData(cSess, pl)
-}
-
-func payloadInData(cSess *sessdata.ConnSession, payload *sessdata.Payload) bool {
+func payloadIn(cSess *sessdata.ConnSession, pl *sessdata.Payload) bool {
 	// 进行Acl规则判断
-	check := checkLinkAcl(cSess.Group, payload)
+	check := checkLinkAcl(cSess.Group, pl)
 	if !check {
 		// 校验不通过直接丢弃
 		return false
@@ -25,7 +16,7 @@ func payloadInData(cSess *sessdata.ConnSession, payload *sessdata.Payload) bool

 	closed := false
 	select {
-	case cSess.PayloadIn <- payload:
+	case cSess.PayloadIn <- pl:
 	case <-cSess.CloseChan:
 		closed = true
 	}
@@ -33,21 +24,16 @@ func payloadInData(cSess *sessdata.ConnSession, payload *sessdata.Payload) bool
 	return closed
 }

-func payloadOut(cSess *sessdata.ConnSession, lType sessdata.LType, pType byte, data []byte) bool {
+func payloadOut(cSess *sessdata.ConnSession, pl *sessdata.Payload) bool {
 	dSess := cSess.GetDtlsSession()
 	if dSess == nil {
-		return payloadOutCstp(cSess, lType, pType, data)
+		return payloadOutCstp(cSess, pl)
 	} else {
-		return payloadOutDtls(cSess, dSess, lType, pType, data)
+		return payloadOutDtls(cSess, dSess, pl)
 	}
 }

-func payloadOutCstp(cSess *sessdata.ConnSession, lType sessdata.LType, pType byte, data []byte) bool {
-	pl := getPayload()
-	pl.LType = lType
-	pl.PType = pType
-	pl.Data = append(pl.Data, data...)
-
+func payloadOutCstp(cSess *sessdata.ConnSession, pl *sessdata.Payload) bool {
 	closed := false

 	select {
@@ -59,12 +45,7 @@ func payloadOutCstp(cSess *sessdata.ConnSession, lType sessdata.LType, pType byt
 	return closed
 }

-func payloadOutDtls(cSess *sessdata.ConnSession, dSess *sessdata.DtlsSession, lType sessdata.LType, pType byte, data []byte) bool {
-	pl := getPayload()
-	pl.LType = lType
-	pl.PType = pType
-	pl.Data = append(pl.Data, data...)
-
+func payloadOutDtls(cSess *sessdata.ConnSession, dSess *sessdata.DtlsSession, pl *sessdata.Payload) bool {
 	select {
 	case cSess.PayloadOutDtls <- pl:
 	case <-dSess.CloseChan:
@@ -74,15 +55,16 @@ func payloadOutDtls(cSess *sessdata.ConnSession, dSess *sessdata.DtlsSession, lT
 }

 // Acl规则校验
-func checkLinkAcl(group *dbdata.Group, payload *sessdata.Payload) bool {
-	if payload.LType == sessdata.LTypeIPData && payload.PType == 0x00 && len(group.LinkAcl) > 0 {
+func checkLinkAcl(group *dbdata.Group, pl *sessdata.Payload) bool {
+	if pl.LType == sessdata.LTypeIPData && pl.PType == 0x00 && len(group.LinkAcl) > 0 {
 	} else {
 		return true
 	}

-	ip_dst := waterutil.IPv4Destination(payload.Data)
-	ip_port := waterutil.IPv4DestinationPort(payload.Data)
-	ip_proto := waterutil.IPv4Protocol(payload.Data)
+	data := pl.Data
+	ip_dst := waterutil.IPv4Destination(data)
+	ip_port := waterutil.IPv4DestinationPort(data)
+	ip_proto := waterutil.IPv4Protocol(data)
 	// fmt.Println("sent:", ip_dst, ip_port)

 	// 优先放行dns端口
--- a/server/handler/pool.go
+++ b/server/handler/pool.go
@@ -3,13 +3,21 @@ package handler
 import (
 	"sync"

+	"github.com/bjdgyc/anylink/base"
 	"github.com/bjdgyc/anylink/sessdata"
 )

+// 不允许直接修改
+// [6] => PType
+var plHeader = []byte{'S', 'T', 'F', 0x01, 0x00, 0x00, 0x00, 0x00}
+
 var plPool = sync.Pool{
 	New: func() interface{} {
+		b := make([]byte, BufferSize)
 		pl := sessdata.Payload{
-			Data: make([]byte, 0, BufferSize),
+			LType: sessdata.LTypeIPData,
+			PType: 0x00,
+			Data:  b,
 		}
 		// fmt.Println("plPool-init", len(pl.Data), cap(pl.Data))
 		return &pl
@@ -22,15 +30,21 @@ func getPayload() *sessdata.Payload {
 }

 func putPayload(pl *sessdata.Payload) {
-	pl.LType = 0
-	pl.PType = 0
-	pl.Data = pl.Data[:0]
+	// 错误数据丢弃
+	if cap(pl.Data) != BufferSize {
+		base.Warn("payload cap is err", cap(pl.Data))
+		return
+	}
+
+	pl.LType = sessdata.LTypeIPData
+	pl.PType = 0x00
+	pl.Data = pl.Data[:BufferSize]
 	plPool.Put(pl)
 }

 var bytePool = sync.Pool{
 	New: func() interface{} {
-		b := make([]byte, 0, BufferSize)
+		b := make([]byte, BufferSize)
 		// fmt.Println("bytePool-init")
 		return &b
 	},
@@ -38,15 +52,15 @@ var bytePool = sync.Pool{

 func getByteZero() *[]byte {
 	b := bytePool.Get().(*[]byte)
+	*b = (*b)[:0]
 	return b
 }

 func getByteFull() *[]byte {
 	b := bytePool.Get().(*[]byte)
-	*b = (*b)[:BufferSize]
 	return b
 }
 func putByte(b *[]byte) {
-	*b = (*b)[:0]
+	*b = (*b)[:BufferSize]
 	bytePool.Put(b)
 }