Add agent gRPC plugin

Add plugin to send pcap data over Unix socket using gRPC

.gitmodules

@@ -0,0 +1,4 @@
+[submodule "deps/agent-plugins-grpc"]
+	path = deps/agent-plugins-grpc
+	url = [email protected]:deepfence/agent-plugins-grpc.git
+	branch = add-pcap

Makefile

@@ -13,9 +13,12 @@ ifeq ($(RELEASE),1)
 	LDFLAGS += -s -w
 endif
 
-.PHONY: all build docker-bin docker-image test
+.PHONY: all build docker-bin docker-image test localinit
 
-all: build
+all: proto build
+
+localinit:
+	$(PWD)/bootstrap.sh
 
 build:
 	go build -tags '$(TAGS)' --ldflags '$(LDFLAGS)' -o packetstreamer ./main.go
@@ -35,3 +38,13 @@ docker-test:
 
 test:
 	go test -tags '$(TAGS)' ./...
+
+clean:
+	-rm ./packetstreamer
+	-rm ./deps/agent-plugins-grpc/proto/*.go
+	-rm -r $(PWD)/proto
+
+proto: ./deps/agent-plugins-grpc/proto/*.proto
+	(cd ./deps/agent-plugins-grpc && make go)
+	-mkdir $(PWD)/proto
+	cp ./deps/agent-plugins-grpc/proto/*.go $(PWD)/proto

README.md

@@ -33,7 +33,7 @@ little performance impact on the remote hosts. PacketStreamer sensors can be
 run on bare-metal servers, on Docker hosts, and on Kubernetes nodes.
 
 The PacketStreamer receiver accepts network traffic from multiple sensors,
-collecting it into a single, central `pcap` file.  You can then process the 
+collecting it into a single, central `pcap` file.  You can then process the
 pcap file or live feed the traffic to the tooling of your choice, such as
 `Zeek`, `Wireshark` `Suricata`, or as a live stream for Machine Learning models.
 
@@ -54,16 +54,17 @@ network data from multiple machines for central logging and analysis.
 For full instructions, refer to the [PacketStreamer Documentation](https://deepfence.github.io/PacketStreamer/).
 
 You will need to install the golang toolchain and `libpcap-dev` before building PacketStreamer.
-  
+
 ```shell script
 # Pre-requisites (Ubuntu): sudo apt install golang-go libpcap-dev
 git clone https://github.com/deepfence/PacketStreamer.git
 cd PacketStreamer/
+make localinit
 make
 ```
 
 Run a PacketStreamer receiver, listening on port **8081** and writing pcap output to **/tmp/dump_file** (see [receiver.yaml](contrib/config/receiver.yaml)):
-  
+
 ```shell script
 ./packetstreamer receiver --config ./contrib/config/receiver.yaml
 ```
@@ -79,7 +80,7 @@ cp ./contrib/config/sensor-local.yaml ./contrib/config/sensor.yaml
 ./packetstreamer sensor --config ./contrib/config/sensor.yaml
 ```
 
-  
+
 ## Who uses PacketStreamer?
 
  * Deepfence [ThreatStryker](https://deepfence.io/threatstryker/) uses

bootstrap.sh

@@ -0,0 +1 @@
+git submodule update --init --remote --recursive ./deps/agent-plugins-grpc/

contrib/config/sensor-agent.yaml

@@ -0,0 +1,5 @@
+output:
+  plugins:
+    agent:
+      socketPath: /tmp/agent.sock
+pcapMode: all

go.mod

@@ -16,6 +16,13 @@ require (
 	gopkg.in/yaml.v3 v3.0.1
 )
 
+require (
+	github.com/golang/protobuf v1.5.2 // indirect
+	golang.org/x/text v0.3.6 // indirect
+	google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013 // indirect
+	google.golang.org/protobuf v1.27.1 // indirect
+)
+
 require (
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.4.1 // indirect
 	github.com/aws/aws-sdk-go-v2/credentials v1.11.2 // indirect
@@ -35,8 +42,9 @@ require (
 	github.com/miekg/dns v1.1.25 // indirect
 	github.com/pierrec/lz4/v4 v4.1.14 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
-	golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550 // indirect
+	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 // indirect
 	golang.org/x/net v0.0.0-20211209124913-491a49abca63 // indirect
 	golang.org/x/sys v0.0.0-20211210111614-af8b64212486 // indirect
+	google.golang.org/grpc v1.48.0
 	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
 )

pkg/config/config.go

@@ -64,9 +64,14 @@ type KafkaPluginConfig struct {
 	Timeout     time.Duration      `yaml:"timeout,omitempty"`
 }
 
+type AgentPluginConfig struct {
+	SocketPath string `yaml:"socketPath"`
+}
+
 type PluginsConfig struct {
 	S3    *S3PluginConfig
 	Kafka *KafkaPluginConfig
+	Agent *AgentPluginConfig
 }
 
 type OutputConfig struct {
@@ -75,6 +80,10 @@ type OutputConfig struct {
 	Plugins *PluginsConfig
 }
 
+type AgentOutputRawConfig struct {
+	SocketPath string `yaml:"socketPath"`
+}
+
 type S3OutputRawConfig struct {
 	Bucket          string
 	Region          string
@@ -97,6 +106,7 @@ type KafkaOutputRawConfig struct {
 type PluginsRawConfig struct {
 	S3    *S3OutputRawConfig
 	Kafka *KafkaOutputRawConfig
+	Agent *AgentOutputRawConfig
 }
 
 type OutputRawConfig struct {
@@ -166,6 +176,7 @@ func NewConfig(configFileName string) (*Config, error) {
 
 	var s3Config *S3PluginConfig
 	var kafkaConfig *KafkaPluginConfig
+	var agentConfig *AgentPluginConfig
 	if rawConfig.Output != nil && rawConfig.Output.Plugins != nil {
 
 		s3Config, err = populateS3Config(rawConfig)
@@ -179,6 +190,12 @@ func NewConfig(configFileName string) (*Config, error) {
 		if err != nil {
 			return nil, err
 		}
+
+		agentConfig, err = populateAgentConfig(rawConfig)
+
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	compressBlockSize := 65
@@ -213,6 +230,7 @@ func NewConfig(configFileName string) (*Config, error) {
 			Plugins: &PluginsConfig{
 				S3:    s3Config,
 				Kafka: kafkaConfig,
+				Agent: agentConfig,
 			},
 		},
 		TLS:                    rawConfig.TLS,
@@ -303,6 +321,16 @@ func populateKafkaConfig(rawConfig RawConfig) (*KafkaPluginConfig, error) {
 	}, nil
 }
 
+func populateAgentConfig(rawConfig RawConfig) (*AgentPluginConfig, error) {
+	if rawConfig.Output.Plugins.Agent == nil {
+		return nil, nil
+	}
+
+	return &AgentPluginConfig{
+		SocketPath: rawConfig.Output.Plugins.Agent.SocketPath,
+	}, nil
+}
+
 func populateS3Config(rawConfig RawConfig) (*S3PluginConfig, error) {
 	if rawConfig.Output.Plugins.S3 == nil {
 		return nil, nil

pkg/config/sensor.go

@@ -12,7 +12,7 @@ var (
 func ValidateSensorConfig(config *Config) error {
 	if config.Output.File == nil && config.Output.Server == nil &&
 		(config.Output.Plugins == nil ||
-			(config.Output.Plugins.S3 == nil && config.Output.Plugins.Kafka == nil)) {
+			(config.Output.Plugins.S3 == nil && config.Output.Plugins.Kafka == nil && config.Output.Plugins.Agent == nil)) {
 		return ErrNoOutputConfigured
 	}
 	if config.Output.Server != nil && config.Output.Server.Port == nil {

pkg/plugins/agent/agent.go

@@ -0,0 +1,76 @@
+package agent
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/deepfence/PacketStreamer/pkg/config"
+	pb "github.com/deepfence/PacketStreamer/proto"
+	"google.golang.org/grpc"
+)
+
+const (
+	SOCKET_RETRIES = 10
+)
+
+type Plugin struct {
+	client   pb.PcapForwarderClient
+	stopChan chan struct{}
+}
+
+func NewPlugin(conf *config.AgentPluginConfig) (*Plugin, error) {
+
+	_, errw := os.Stat(conf.SocketPath)
+	tries := 1
+	for errors.Is(errw, os.ErrNotExist) {
+		_, errw = os.Stat(conf.SocketPath)
+		time.Sleep(1 * time.Second)
+		if tries == SOCKET_RETRIES {
+			return nil, fmt.Errorf("No socket found: %v", errw)
+		}
+		tries += 1
+	}
+
+	conn, err := grpc.Dial("unix://"+conf.SocketPath, grpc.WithAuthority("dummy"), grpc.WithInsecure())
+	if err != nil {
+		return nil, err
+	}
+
+	return &Plugin{
+		client:   pb.NewPcapForwarderClient(conn),
+		stopChan: make(chan struct{}, 1),
+	}, nil
+}
+
+func (p *Plugin) Start(ctx context.Context) (chan<- string, error) {
+
+	stream, err := p.client.SendPackets(ctx)
+
+	if err != nil {
+		return nil, err
+	}
+
+	recvChan := make(chan string, 10)
+
+	go func() {
+	loop:
+		for {
+			select {
+			case packet := <-recvChan:
+				stream.Send(&pb.Packet{Payload: []byte(packet)})
+			case <-p.stopChan:
+				break loop
+			}
+		}
+		stream.CloseAndRecv()
+	}()
+
+	return recvChan, nil
+}
+
+func (p *Plugin) Stop() {
+	p.stopChan <- struct{}{}
+}

pkg/plugins/plugins.go

@@ -6,6 +6,7 @@ import (
 	"log"
 
 	"github.com/deepfence/PacketStreamer/pkg/config"
+	"github.com/deepfence/PacketStreamer/pkg/plugins/agent"
 	"github.com/deepfence/PacketStreamer/pkg/plugins/kafka"
 	"github.com/deepfence/PacketStreamer/pkg/plugins/s3"
 )
@@ -43,6 +44,23 @@ func Start(ctx context.Context, config *config.Config) (chan<- string, error) {
 		plugins = append(plugins, kafkaChan)
 	}
 
+	if config.Output.Plugins.Agent != nil {
+		log.Println("Starting Agent plugin")
+		agentPlugin, err := agent.NewPlugin(config.Output.Plugins.Agent)
+
+		if err != nil {
+			return nil, fmt.Errorf("error starting Agent plugin, %v", err)
+		}
+
+		agentChan, err := agentPlugin.Start(ctx)
+
+		if err != nil {
+			return nil, fmt.Errorf("error starting Agent plugin, %v", err)
+		}
+
+		plugins = append(plugins, agentChan)
+	}
+
 	inputChan := make(chan string)
 	go func() {
 		defer func() {
@@ -66,5 +84,5 @@ func Start(ctx context.Context, config *config.Config) (chan<- string, error) {
 }
 
 func pluginsAreDefined(pluginsConfig *config.PluginsConfig) bool {
-	return pluginsConfig != nil && (pluginsConfig.S3 != nil || pluginsConfig.Kafka != nil)
+	return pluginsConfig != nil && (pluginsConfig.Agent != nil || pluginsConfig.S3 != nil || pluginsConfig.Kafka != nil)
 }