Support send traffic to bubblereplay

.gitignore

@@ -5,6 +5,9 @@
 .vscode/
 .idea/
 
+# IDL
+idl/
+
 # Binaries for programs and plugins
 bubblecopy
 output/

Makefile

@@ -2,13 +2,15 @@
 define DEBUG_SETTINGS
 {
 	"interface": "lo0",
-	"service_port": "8080"
+	"service_port": "8080",
+	"replay_svr_addr": "127.0.0.1:6789"
 }
 endef
 export DEBUG_SETTINGS
 
 # todo: cross-compile for linux and windows to releases.
 build: clean
+	@echo "go build project..."
 	@mkdir -p output
 	@go build -o output/bubblecopy
 
@@ -23,4 +25,14 @@ run-debug: build
 
 clean:
 	@echo "clean output directory..."
-	@rm -rf output/
+	@rm -rf output/
+
+update-idl:
+	@rm -rf ./idl
+	@echo "step1> fetching idl repo..."
+	@git clone --depth=1 https://github.com/bubble-diff/IDL.git idl
+	@rm -rf ./idl/.git
+	@rm ./idl/.gitignore
+	@echo "step2> compile idl..."
+	@protoc --go_out=. idl/*.proto
+	@go mod tidy

buffer.go

@@ -1,14 +1,18 @@
 package main
 
-import "io"
+import (
+	"io"
+)
 
 type buffer struct {
 	bytes chan []byte
 	data  []byte
 }
 
 func NewBuffer() *buffer {
-	// todo: bytes可以做成带缓存的channel吗？
+	// 这里，必须是无缓存的channel，因为channel是stream进行close的。
+	// 如果带缓存，stream关掉channel后，consumer会消费失败。
+	// todo: 将close交给consume去做？这样就可以带缓存了
 	return &buffer{bytes: make(chan []byte)}
 }
 

config.go

@@ -10,13 +10,15 @@ import (
 
 type config struct {
 	// Taskid Diff任务ID
-	Taskid int `json:"taskid"`
+	Taskid int64 `json:"taskid"`
 	// Secret 访问对应id任务配置的密钥
 	Secret string `json:"secret"`
 	// Device 网卡名称
 	Device string `json:"interface"`
 	// Port 被测服务端口
 	Port string `json:"service_port"`
+	// ReplaySvrAddr bubblereplay服务地址
+	ReplaySvrAddr string `json:"replay_svr_addr"`
 
 	// DeviceIPv4 网卡ipv4地址
 	DeviceIPv4 string
@@ -35,6 +37,10 @@ func (c *config) init() {
 		logrus.Fatal(err)
 	}
 
+	if configuration.ReplaySvrAddr == "" {
+		logrus.Fatal("bubblereplay server addr not set")
+	}
+
 	c.DeviceIPv4, err = getDeviceIpv4(c.Device)
 	if err != nil {
 		logrus.Error(err)

const.go

@@ -11,3 +11,7 @@ const (
 	UnknownType = "unknown"
 	HttpType    = "http"
 )
+
+const (
+	ApiAddRecord = "/record/add"
+)

go.mod

@@ -5,6 +5,7 @@ go 1.17
 require (
 	github.com/google/gopacket v1.1.19
 	github.com/sirupsen/logrus v1.8.1
+	google.golang.org/protobuf v1.27.1
 )
 
 require golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27 // indirect

go.sum

@@ -1,5 +1,8 @@
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
 github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -24,3 +27,8 @@ golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.27.1 h1:SnqbnDw1V7RiZcXPx5MEeqPv2s79L9i7BJUlG/+RurQ=
+google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=

main.go

@@ -4,6 +4,7 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"os/signal"
 	"time"
 
 	"github.com/google/gopacket"
@@ -61,14 +62,19 @@ func main() {
 	streamPool := reassembly.NewStreamPool(streamFactory)
 	assembler := reassembly.NewAssembler(streamPool)
 
+	signalChan := make(chan os.Signal, 1)
+	signal.Notify(signalChan, os.Interrupt)
+
 	ticker := time.NewTicker(time.Second * 30)
-	defer ticker.Stop()
-	defer streamFactory.WaitConsumers()
 	for {
 		// todo: 等待Diff任务启动，若未启动，请勿进行抓包消耗CPU
 		// your code here...
 
+		done := false
 		select {
+		case <-signalChan:
+			logrus.Info("Caught SIGINT: aborting")
+			done = true
 		case <-ticker.C:
 			// 停止监听30秒内无数据传输的连接
 			assembler.FlushCloseOlderThan(time.Now().Add(time.Second * -30))
@@ -79,5 +85,16 @@ func main() {
 				assembler.Assemble(packet.NetworkLayer().NetworkFlow(), tcp)
 			}
 		}
+		if done {
+			break
+		}
 	}
+
+	ticker.Stop()
+	// Important! Please flush all connection before waiting consumers.
+	closed := assembler.FlushAll()
+	logrus.Debugf("Final flush: %d closed", closed)
+
+	streamFactory.WaitConsumers()
+	logrus.Info("Bye~")
 }