添加ip审计功能

Dockerfile

@@ -2,22 +2,21 @@
 FROM node:lts-alpine as builder_node
 WORKDIR /web
 COPY ./web /web
-RUN npx browserslist@latest --update-db \
+RUN npm install --registry=https://registry.npm.taobao.org \
-    && npm install \
     && npm run build \
     && ls /web/ui
 
 # server
 FROM golang:1.16-alpine as builder_golang
 #TODO 本地打包时使用镜像
-#ENV GOPROXY=https://goproxy.io
+ENV GOPROXY=https://goproxy.io
 ENV GOOS=linux
 WORKDIR /anylink
 COPY . /anylink
 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 sed -i 's/dl-cdn.alpinelinux.org/mirrors.tuna.tsinghua.edu.cn/g' /etc/apk/repositories
 RUN apk add --no-cache git gcc musl-dev
 RUN cd /anylink/server;go build -o anylink -ldflags "-X main.CommitId=$(git rev-parse HEAD)" \
     && /anylink/server/anylink tool -v
@@ -37,7 +36,7 @@ COPY ./server/files  /app/conf/files
 
 
 #TODO 本地打包时使用镜像
-#RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.tuna.tsinghua.edu.cn/g' /etc/apk/repositories
+RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.tuna.tsinghua.edu.cn/g' /etc/apk/repositories
 RUN apk add --no-cache bash iptables \
     && chmod +x /app/docker_entrypoint.sh \
     && ls /app

docker_build.sh

@@ -0,0 +1,13 @@
+#!/bin/env bash
+
+ver="0.5.1"
+
+#docker login -u bjdgyc
+
+docker build -t bjdgyc/anylink .
+
+docker tag bjdgyc/anylink:latest bjdgyc/anylink:$ver
+
+docker push bjdgyc/anylink:$ver
+docker push bjdgyc/anylink:latest
+

server/base/app_ver.go

@@ -2,6 +2,6 @@ package base
 
 const (
 	APP_NAME = "AnyLink"
-	// 修改为sql数据库
+	// 添加访问日志审计
-	APP_VER = "0.5.1"
+	APP_VER = "0.5.2"
 )

server/base/cfg.go

@@ -65,6 +65,7 @@ type ServerConfig struct {
 
 	SessionTimeout int `json:"session_timeout"` // in seconds
 	// AuthTimeout    int `json:"auth_timeout"`    // in seconds
+	AuditInterval int `json:"audit_interval"` // in seconds
 }
 
 func initServerCfg() {

server/base/config.go

@@ -54,6 +54,7 @@ var configs = []config{
 	{Typ: cfgInt, Name: "mobile_dpd", Usage: "移动端死链接检测时间(秒)", ValInt: 60},
 	{Typ: cfgInt, Name: "session_timeout", Usage: "session过期时间(秒)", ValInt: 3600},
 	// {Typ: cfgInt, Name: "auth_timeout", Usage: "auth_timeout", ValInt: 0},
+	{Typ: cfgInt, Name: "audit_interval", Usage: "审计去重间隔(秒)", ValInt: 1800},
 }
 
 var envs = map[string]string{}

server/dbdata/db.go

@@ -25,7 +25,7 @@ func initDb() {
 	}
 
 	// 初始化数据库
-	err = xdb.Sync2(&User{}, &Setting{}, &Group{}, &IpMap{})
+	err = xdb.Sync2(&User{}, &Setting{}, &Group{}, &IpMap{}, &AccessAudit{})
 	if err != nil {
 		base.Fatal(err)
 	}

server/dbdata/tables.go

@@ -54,3 +54,14 @@ type Setting struct {
 	Data      json.RawMessage `json:"data" xorm:"Text"`
 	UpdatedAt time.Time       `json:"updated_at" xorm:"DateTime updated"`
 }
+
+type AccessAudit struct {
+	Id        int       `json:"id" xorm:"pk autoincr not null"`
+	Username  string    `json:"username" xorm:"varchar(60) not null"`
+	Protocol  uint8     `json:"protocol" xorm:"not null"`
+	Src       string    `json:"src" xorm:"varchar(60) not null"`
+	SrcPort   uint16    `json:"src_port" xorm:"not null"`
+	Dst       string    `json:"dst" xorm:"varchar(60) not null"`
+	DstPort   uint16    `json:"dst_port" xorm:"not null"`
+	CreatedAt time.Time `json:"created_at" xorm:"DateTime"`
+}

server/handler/link_cstp.go

@@ -1,6 +1,7 @@
 package handler
 
 import (
+	"bufio"
 	"encoding/binary"
 	"net"
 	"time"
@@ -9,7 +10,7 @@ import (
 	"github.com/bjdgyc/anylink/sessdata"
 )
 
-func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
+func LinkCstp(conn net.Conn, bufRW *bufio.ReadWriter, cSess *sessdata.ConnSession) {
 	defer func() {
 		base.Debug("LinkCstp return", cSess.IpAddr)
 		_ = conn.Close()
@@ -23,7 +24,7 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
 		dead    = time.Duration(cSess.CstpDpd+5) * time.Second
 	)
 
-	go cstpWrite(conn, cSess)
+	go cstpWrite(conn, bufRW, cSess)
 
 	for {
 
@@ -35,7 +36,7 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
 		}
 		// hdata := make([]byte, BufferSize)
 		pl := getPayload()
-		n, err = conn.Read(pl.Data)
+		n, err = bufRW.Read(pl.Data)
 		if err != nil {
 			base.Error("read hdata: ", err)
 			return
@@ -77,7 +78,7 @@ func LinkCstp(conn net.Conn, cSess *sessdata.ConnSession) {
 	}
 }
 
-func cstpWrite(conn net.Conn, cSess *sessdata.ConnSession) {
+func cstpWrite(conn net.Conn, bufRW *bufio.ReadWriter, cSess *sessdata.ConnSession) {
 	defer func() {
 		base.Debug("cstpWrite return", cSess.IpAddr)
 		_ = conn.Close()

server/handler/link_tunnel.go

@@ -147,7 +147,7 @@ func LinkTunnel(w http.ResponseWriter, r *http.Request) {
 	base.Debug(buf.String())
 
 	hj := w.(http.Hijacker)
-	conn, _, err := hj.Hijack()
+	conn, bufRW, err := hj.Hijack()
 	if err != nil {
 		base.Error(err)
 		w.WriteHeader(http.StatusInternalServerError)
@@ -166,5 +166,5 @@ func LinkTunnel(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	go LinkCstp(conn, cSess)
+	go LinkCstp(conn, bufRW, cSess)
 }

server/handler/payload.go

@@ -1,17 +1,25 @@
 package handler
 
 import (
+	"encoding/binary"
+
+	"github.com/bjdgyc/anylink/base"
 	"github.com/bjdgyc/anylink/dbdata"
+	"github.com/bjdgyc/anylink/pkg/utils"
 	"github.com/bjdgyc/anylink/sessdata"
 	"github.com/songgao/water/waterutil"
 )
 
 func payloadIn(cSess *sessdata.ConnSession, pl *sessdata.Payload) bool {
-	// 进行Acl规则判断
+	if pl.LType == sessdata.LTypeIPData && pl.PType == 0x00 {
-	check := checkLinkAcl(cSess.Group, pl)
+		// 进行Acl规则判断
-	if !check {
+		check := checkLinkAcl(cSess.Group, pl)
-		// 校验不通过直接丢弃
+		if !check {
-		return false
+			// 校验不通过直接丢弃
+			return false
+		}
+
+		logAudit(cSess, pl)
 	}
 
 	closed := false
@@ -61,24 +69,23 @@ func checkLinkAcl(group *dbdata.Group, pl *sessdata.Payload) bool {
 		return true
 	}
 
-	data := pl.Data
+	ipDst := waterutil.IPv4Destination(pl.Data)
-	ip_dst := waterutil.IPv4Destination(data)
+	ipPort := waterutil.IPv4DestinationPort(pl.Data)
-	ip_port := waterutil.IPv4DestinationPort(data)
+	ipProto := waterutil.IPv4Protocol(pl.Data)
-	ip_proto := waterutil.IPv4Protocol(data)
 	// fmt.Println("sent:", ip_dst, ip_port)
 
 	// 优先放行dns端口
 	for _, v := range group.ClientDns {
-		if v.Val == ip_dst.String() && ip_port == 53 {
+		if v.Val == ipDst.String() && ipPort == 53 {
 			return true
 		}
 	}
 
 	for _, v := range group.LinkAcl {
 		// 循环判断ip和端口
-		if v.IpNet.Contains(ip_dst) {
+		if v.IpNet.Contains(ipDst) {
 			// 放行允许ip的ping
-			if v.Port == ip_port || v.Port == 0 || ip_proto == waterutil.ICMP {
+			if v.Port == ipPort || v.Port == 0 || ipProto == waterutil.ICMP {
 				if v.Action == dbdata.Allow {
 					return true
 				} else {
@@ -90,3 +97,49 @@ func checkLinkAcl(group *dbdata.Group, pl *sessdata.Payload) bool {
 
 	return false
 }
+
+// 访问日志审计
+func logAudit(cSess *sessdata.ConnSession, pl *sessdata.Payload) {
+	ipSrc := waterutil.IPv4Source(pl.Data)
+	ipDst := waterutil.IPv4Destination(pl.Data)
+	ipPort := waterutil.IPv4DestinationPort(pl.Data)
+	ipProto := waterutil.IPv4Protocol(pl.Data)
+
+	// 只统计 tcp和udp 的访问
+	switch ipProto {
+	case waterutil.TCP:
+	case waterutil.UDP:
+	default:
+		return
+	}
+
+	b := getByte34()
+	key := *b
+	copy(key[:16], ipSrc)
+	copy(key[16:32], ipDst)
+	binary.BigEndian.PutUint16(key[32:34], ipPort)
+
+	s := utils.BytesToString(key)
+	nu := utils.NowSec().Unix()
+
+	// 判断已经存在，并且没有过期
+	v, ok := cSess.IpAuditMap[s]
+	if ok && nu-v < int64(base.Cfg.AuditInterval) {
+		// 回收byte对象
+		putByte34(b)
+		return
+	}
+
+	cSess.IpAuditMap[s] = nu
+
+	audit := dbdata.AccessAudit{
+		Username:  cSess.Sess.Username,
+		Protocol:  uint8(ipProto),
+		Src:       ipSrc.String(),
+		Dst:       ipDst.String(),
+		DstPort:   ipPort,
+		CreatedAt: utils.NowSec(),
+	}
+
+	_ = dbdata.Add(audit)
+}

server/handler/pool.go

@@ -9,7 +9,12 @@ import (
 
 // 不允许直接修改
 // [6] => PType
-var plHeader = []byte{'S', 'T', 'F', 0x01, 0x00, 0x00, 0x00, 0x00}
+var plHeader = []byte{
+	'S', 'T', 'F', 1,
+	0x00, 0x00, /* Length */
+	0x00, /* Type */
+	0x00, /* Unknown */
+}
 
 var plPool = sync.Pool{
 	New: func() interface{} {
@@ -64,3 +69,21 @@ func putByte(b *[]byte) {
 	*b = (*b)[:BufferSize]
 	bytePool.Put(b)
 }
+
+// 长度 34 小对象
+var byte34Pool = sync.Pool{
+	New: func() interface{} {
+		b := make([]byte, 34)
+		return &b
+	},
+}
+
+func getByte34() *[]byte {
+	b := byte34Pool.Get().(*[]byte)
+	return b
+}
+
+func putByte34(b *[]byte) {
+	*b = (*b)[:34]
+	byte34Pool.Put(b)
+}

server/handler/server.go

@@ -10,11 +10,10 @@ import (
 	"os"
 	"time"
 
-	"github.com/pion/dtls/v2/pkg/crypto/selfsign"
-
 	"github.com/bjdgyc/anylink/base"
 	"github.com/bjdgyc/anylink/pkg/proxyproto"
 	"github.com/gorilla/mux"
+	"github.com/pion/dtls/v2/pkg/crypto/selfsign"
 )
 
 func startTls() {

server/pkg/utils/unsafe.go

@@ -0,0 +1,20 @@
+package utils
+
+import (
+	"unsafe"
+)
+
+// BytesToString converts byte slice to string.
+func BytesToString(b []byte) string {
+	return *(*string)(unsafe.Pointer(&b))
+}
+
+// StringToBytes converts string to byte slice.
+func StringToBytes(s string) []byte {
+	return *(*[]byte)(unsafe.Pointer(
+		&struct {
+			string
+			Cap int
+		}{s, len(s)},
+	))
+}

server/pkg/utils/util.go

@@ -3,11 +3,30 @@ package utils
 import (
 	"fmt"
 	"math/rand"
+	"sync/atomic"
 	"time"
 )
 
+var (
+	// 每秒时间缓存
+	timeNowSec = &atomic.Value{}
+)
+
 func init() {
 	rand.Seed(time.Now().UnixNano())
+
+	timeNowSec.Store(time.Now())
+	go func() {
+		tick := time.NewTicker(time.Second * 1)
+		for c := range tick.C {
+			timeNowSec.Store(c)
+		}
+	}()
+}
+
+func NowSec() time.Time {
+	t := timeNowSec.Load()
+	return t.(time.Time)
 }
 
 func InArrStr(arr []string, str string) bool {

server/sessdata/session.go

@@ -46,9 +46,9 @@ type ConnSession struct {
 	closeOnce           sync.Once
 	CloseChan           chan struct{}
 	PayloadIn           chan *Payload
-	// PayloadOut          chan *Payload // 公共ip数据
+	PayloadOutCstp      chan *Payload    // Cstp的数据
-	PayloadOutCstp chan *Payload // Cstp的数据
+	PayloadOutDtls      chan *Payload    // Dtls的数据
-	PayloadOutDtls chan *Payload // Dtls的数据
+	IpAuditMap          map[string]int64 // 审计的ip数据
 
 	// dSess *DtlsSession
 	dSess *atomic.Value
@@ -186,6 +186,7 @@ func (s *Session) NewConn() *ConnSession {
 		PayloadIn:      make(chan *Payload, 64),
 		PayloadOutCstp: make(chan *Payload, 64),
 		PayloadOutDtls: make(chan *Payload, 64),
+		IpAuditMap:     make(map[string]int64, 512),
 		dSess:          &atomic.Value{},
 	}