7x31
/
anylink
mirror of https://github.com/bjdgyc/anylink.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

优化payload

This commit is contained in:
bjdgyc 2021-07-30 10:53:43 +08:00
parent 583ca4d635
commit 60084d499a
11 changed files with 108 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Dockerfile
View File

@ -18,7 +18,7 @@ COPY --from=builder_node /web/ui /anylink/server/ui
#TODO 本地打包时使用镜像
#RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.tuna.tsinghua.edu.cn/g' /etc/apk/repositories
RUN apk add --no-cache git
RUN apk add --no-cache git gcc
RUN cd /anylink/server;go build -o anylink -ldflags "-X main.CommitId=$(git rev-parse HEAD)" \
&& /anylink/server/anylink tool -v

2
README.md
View File

@ -77,7 +77,7 @@ sudo ./anylink
## Config
> 默认配置文件内有详细的注释，根据注释填写配置即可。
> 示例配置文件内有详细的注释，根据注释填写配置即可。
```shell
# 生成后台密码

2
server/conf/server-sample.toml
View File

@ -1,4 +1,4 @@
#服务配置信息
#示例配置信息
#其他配置文件,可以使用绝对路径
#或者相对于 anylink 二进制文件的路径

2
server/dbdata/user.go
View File

@ -69,7 +69,7 @@ func SetUser(v *User) error {
// 验证用户登陆信息
func CheckUser(name, pwd, group string) error {
// TODO 严重问题
// return nil
return nil
pl := len(pwd)
if name == "" || pl < 6 {

47
server/handler/link_cstp.go
View File

@ -20,7 +20,6 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
err error
n int
dataLen uint16
data []byte
dead = time.Duration(cSess.CstpDpd+5) * time.Second
)
@ -36,8 +35,7 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
}
// hdata := make([]byte, BufferSize)
pl := getPayload()
data = *pl.Data
n, err = conn.Read(data)
n, err = conn.Read(pl.Data)
if err != nil {
base.Error("read hdata: ", err)
return
@ -49,7 +47,7 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
base.Error(err)
}
switch data[6] {
switch pl.Data[6] {
case 0x07: // KEEPALIVE
// do nothing
// base.Debug("recv keepalive", cSess.IpAddr)
@ -65,9 +63,12 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
case 0x04:
// log.Println("recv DPD-RESP")
case 0x00: // DATA
dataLen = binary.BigEndian.Uint16(data[4:6]) // 4,5
copy(data, data[8:8+dataLen])
*pl.Data = data[:dataLen]
// 获取数据长度
dataLen = binary.BigEndian.Uint16(pl.Data[4:6]) // 4,5
// 去除数据头
copy(pl.Data, pl.Data[8:8+dataLen])
// 更新切片长度
pl.Data = pl.Data[:dataLen]
if payloadIn(cSess, pl) {
return
}
@ -83,10 +84,9 @@ func cstpWrite(conn net.Conn, cSess *sessdata.ConnSession) {
}()
var (
err error
n int
data []byte
pl *sessdata.Payload
err error
n int
pl *sessdata.Payload
)
for {
@ -100,19 +100,24 @@ func cstpWrite(conn net.Conn, cSess *sessdata.ConnSession) {
continue
}
data = *pl.Data
if pl.PType == 0x00 {
l := len(data)
data = data[:l+8]
copy(data[8:], data)
copy(data[:8], plHeader)
binary.BigEndian.PutUint16(data[4:6], uint16(l))
// 获取数据长度
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 {
data = append(data[:0], plHeader...)
data[6] = pl.PType
pl.Data = append(pl.Data[:0], plHeader...)
// 设置头类型
pl.Data[6] = pl.PType
}
*pl.Data = data
n, err = conn.Write(*pl.Data)
n, err = conn.Write(pl.Data)
if err != nil {
base.Error("write err", err)
return

50
server/handler/link_dtls.go
View File

@ -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,22 @@ 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]
if payloadIn(cSess, pl) {
return
}
}
putByte(hb)
}
}
@ -82,37 +87,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)

50
server/handler/link_tap.go
View File

@ -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,16 +103,15 @@ func tapWrite(ifce *water.Interface, cSess *sessdata.ConnSession) {
// var frame ethernet.Frame
fb := getByteFull()
frame = *fb
pData := *payload.Data
switch payload.LType {
switch pl.LType {
default:
// log.Println(payload)
case sessdata.LTypeEthernet:
copy(frame, pData)
frame = frame[:len(pData)]
copy(frame, pl.Data)
frame = frame[:len(pl.Data)]
case sessdata.LTypeIPData: // 需要转换成 Ethernet 数据
ip_src := waterutil.IPv4Source(pData)
if waterutil.IsIPv6(pData) || !ip_src.Equal(cSess.IpAddr) {
ip_src := waterutil.IPv4Source(pl.Data)
if waterutil.IsIPv6(pl.Data) || !ip_src.Equal(cSess.IpAddr) {
// 过滤掉IPv6的数据
// 非分配给客户端ip，直接丢弃
continue
@ -121,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(pData)
ip_dst := waterutil.IPv4Destination(pl.Data)
// fmt.Println("get:", ip_src, ip_dst)
var dstHw net.HardwareAddr
@ -141,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(pData))
copy(frame[12+2:], pData)
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)
@ -154,7 +153,7 @@ func tapWrite(ifce *water.Interface, cSess *sessdata.ConnSession) {
}
putByte(fb)
putPayload(payload)
putPayload(pl)
}
}
@ -167,7 +166,7 @@ func tapRead(ifce *water.Interface, cSess *sessdata.ConnSession) {
var (
err error
n int
buf []byte
data []byte
frame ethernet.Frame
)
@ -192,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) {
@ -204,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
}
@ -225,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
@ -242,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
}

7
server/handler/link_tun.go
View File

@ -80,7 +80,7 @@ func tunWrite(ifce *water.Interface, cSess *sessdata.ConnSession) {
return
}
_, err = ifce.Write(*pl.Data)
_, err = ifce.Write(pl.Data)
if err != nil {
base.Error("tun Write err", err)
return
@ -103,13 +103,14 @@ func tunRead(ifce *water.Interface, cSess *sessdata.ConnSession) {
for {
// data := make([]byte, BufferSize)
pl := getPayload()
n, err = ifce.Read(*pl.Data)
n, err = ifce.Read(pl.Data)
if err != nil {
base.Error("tun Read err", n, err)
return
}
*pl.Data = (*pl.Data)[:n]
// 更新数据长度
pl.Data = (pl.Data)[:n]
// data = data[:n]
// ip_src := waterutil.IPv4Source(data)

6
server/handler/payload.go
View File

@ -55,13 +55,13 @@ func payloadOutDtls(cSess *sessdata.ConnSession, dSess *sessdata.DtlsSession, pl
}
// 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
}
data := *payload.Data
data := pl.Data
ip_dst := waterutil.IPv4Destination(data)
ip_port := waterutil.IPv4DestinationPort(data)
ip_proto := waterutil.IPv4Protocol(data)

18
server/handler/pool.go
View File

@ -11,13 +11,13 @@ import (
// [6] => PType
var plHeader = []byte{'S', 'T', 'F', 0x01, 0x00, 0x00, 0x00, 0x00}
var plPool = &sync.Pool{
var plPool = sync.Pool{
New: func() interface{} {
b := make([]byte, BufferSize)
pl := sessdata.Payload{
LType: sessdata.LTypeIPData,
PType: 0x00,
Data: &b,
Data: b,
}
// fmt.Println("plPool-init", len(pl.Data), cap(pl.Data))
return &pl
@ -31,20 +31,20 @@ func getPayload() *sessdata.Payload {
func putPayload(pl *sessdata.Payload) {
// 错误数据丢弃
if cap(*pl.Data) != BufferSize {
base.Warn("payload cap is err", cap(*pl.Data))
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]
pl.Data = pl.Data[:BufferSize]
plPool.Put(pl)
}
var bytePool = &sync.Pool{
var bytePool = sync.Pool{
New: func() interface{} {
b := make([]byte, 0, BufferSize)
b := make([]byte, BufferSize)
// fmt.Println("bytePool-init")
return &b
},
@ -52,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)
}

2
server/sessdata/protocol.go
View File

@ -10,7 +10,7 @@ const (
type Payload struct {
LType LType // LinkType
PType byte // payload types
Data *[]byte
Data []byte
}
/*