Merge pull request #121 from intelops/dev

Dev

cmd/tarian_detector/main.go

@@ -12,8 +12,15 @@ import (
 	"time"
 
 	"github.com/intelops/tarian-detector/pkg/detector"
-	"github.com/intelops/tarian-detector/pkg/utils"
+	"github.com/intelops/tarian-detector/pkg/k8s"
 	"github.com/intelops/tarian-detector/tarian"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+const (
+	// NotInClusterErrMsg is an error message for when the Kubernetes environment is not detected.
+	NotInClusterErrMsg string = "Kubernetes environment not detected. The Kubernetes context has been disabled."
 )
 
 // main is the entry point of the application. It sets up the necessary components
@@ -27,8 +34,6 @@ func main() {
 	watcher, err := K8Watcher()
 	if err != nil {
 		log.Print(err)
-	} else {
-		watcher.Start()
 	}
 
 	// Initialize Tarian eBPF module
@@ -48,6 +53,7 @@ func main() {
 
 	// Add the eBPF module to the detectors
 	eventsDetector.Add(tarianDetector)
+	eventsDetector.SetPodWatcher(watcher)
 
 	// Start the event detectors and defer their closure
 	err = eventsDetector.Start()
@@ -86,17 +92,7 @@ func main() {
 				continue
 			}
 
-			// Retrieve Kubernetes context based on host process ID
-			k8sCtx, err := GetK8sContext(watcher, e["hostProcessId"].(uint32))
-			if err != nil {
-				// Log the error as the Kubernetes context if an error is
-				e["kubernetes"] = err.Error()
-			} else {
-				// Set the Kubernetes context if no error is encountered
-				e["kubernetes"] = k8sCtx
-			}
-
-			utils.PrintEvent(e, eventsDetector.GetTotalCount())
+			fmt.Println(e)
 		}
 	}()
 
@@ -105,3 +101,18 @@ func main() {
 		time.Sleep(1 * time.Minute)
 	}
 }
+
+// K8Watcher initializes and returns a new PodWatcher for the current Kubernetes cluster.
+func K8Watcher() (*k8s.PodWatcher, error) {
+	// Get the in-cluster configuration.
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		return nil, fmt.Errorf("%v. %s", err, NotInClusterErrMsg)
+	}
+
+	// Create a new Kubernetes client set.
+	clientSet := kubernetes.NewForConfigOrDie(config)
+
+	// Return a new PodWatcher for the current Kubernetes cluster.
+	return k8s.NewPodWatcher(clientSet)
+}

go.mod

@@ -28,19 +28,19 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
-	golang.org/x/exp v0.0.0-20240325151524-a685a6edb6d8 // indirect
-	golang.org/x/net v0.23.0 // indirect
+	golang.org/x/exp v0.0.0-20240409090435-93d18d7e34b8 // indirect
+	golang.org/x/net v0.24.0 // indirect
 	golang.org/x/oauth2 v0.19.0 // indirect
 	golang.org/x/sys v0.19.0 // indirect
-	golang.org/x/term v0.18.0 // indirect
+	golang.org/x/term v0.19.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/klog/v2 v2.120.1 // indirect
-	k8s.io/kube-openapi v0.0.0-20240403164606-bc84c2ddaf99 // indirect
+	k8s.io/kube-openapi v0.0.0-20240411171206-dc4e619f62f3 // indirect
 	k8s.io/utils v0.0.0-20240310230437-4693a0247e57 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect

go.sum

@@ -73,6 +73,8 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/exp v0.0.0-20240325151524-a685a6edb6d8 h1:aAcj0Da7eBAtrTp03QXWvm88pSyOt+UgdZw2BFZ+lEw=
 golang.org/x/exp v0.0.0-20240325151524-a685a6edb6d8/go.mod h1:CQ1k9gNrJ50XIzaKCRR2hssIjF07kZFEiieALBM/ARQ=
+golang.org/x/exp v0.0.0-20240409090435-93d18d7e34b8 h1:ESSUROHIBHg7USnszlcdmjBEwdMj9VUvU+OPk4yl2mc=
+golang.org/x/exp v0.0.0-20240409090435-93d18d7e34b8/go.mod h1:/lliqkxwWAhPjf5oSOIJup2XcqJaw8RGS6k3TGEc7GI=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
@@ -81,6 +83,8 @@ golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.23.0 h1:7EYJ93RZ9vYSZAIb2x3lnuvqO5zneoD6IvWjuhfxjTs=
 golang.org/x/net v0.23.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
+golang.org/x/net v0.24.0 h1:1PcaxkF854Fu3+lvBIx5SYn9wRlBzzcnHZSiaFFAb0w=
+golang.org/x/net v0.24.0/go.mod h1:2Q7sJY5mzlzWjKtYUEXSlBWCdyaioyXzRB2RtU8KVE8=
 golang.org/x/oauth2 v0.19.0 h1:9+E/EZBCbTLNrbN35fHv/a/d/mOBatymz1zbtQrXpIg=
 golang.org/x/oauth2 v0.19.0/go.mod h1:vYi7skDa1x015PmRRYZ7+s1cWyPgrPiSYRe4rnsexc8=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -93,6 +97,8 @@ golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o=
 golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.18.0 h1:FcHjZXDMxI8mM3nwhX9HlKop4C0YQvCVCdwYl2wOtE8=
 golang.org/x/term v0.18.0/go.mod h1:ILwASektA3OnRv7amZ1xhE/KTR+u50pbXfZ03+6Nx58=
+golang.org/x/term v0.19.0 h1:+ThwsDv+tYfnJFhF4L8jITxu1tdTWRTZpdsWgEgjL6Q=
+golang.org/x/term v0.19.0/go.mod h1:2CuTdWZ7KHSQwUzKva0cbMg6q2DMI3Mmxp+gKJbskEk=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
@@ -105,6 +111,7 @@ golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.19.0 h1:tfGCXNR1OsFG+sVdLAitlpjAvD/I6dHDKnYrpEZUHkw=
 golang.org/x/tools v0.19.0/go.mod h1:qoJWxmGSIBmAeriMx19ogtrEPrGtDbPK634QFIcLAhc=
+golang.org/x/tools v0.20.0 h1:hz/CVckiOxybQvFw6h7b/q80NTr9IUQb4s1IIzW7KNY=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
@@ -131,6 +138,8 @@ k8s.io/klog/v2 v2.120.1 h1:QXU6cPEOIslTGvZaXvFWiP9VKyeet3sawzTOvdXb4Vw=
 k8s.io/klog/v2 v2.120.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
 k8s.io/kube-openapi v0.0.0-20240403164606-bc84c2ddaf99 h1:w6nThEmGo9zcL+xH1Tu6pjxJ3K1jXFW+V0u4peqN8ks=
 k8s.io/kube-openapi v0.0.0-20240403164606-bc84c2ddaf99/go.mod h1:yD4MZYeKMBwQKVht279WycxKyM84kkAx2DPrTXaeb98=
+k8s.io/kube-openapi v0.0.0-20240411171206-dc4e619f62f3 h1:SbdLaI6mM6ffDSJCadEaD4IkuPzepLDGlkd2xV0t1uA=
+k8s.io/kube-openapi v0.0.0-20240411171206-dc4e619f62f3/go.mod h1:yD4MZYeKMBwQKVht279WycxKyM84kkAx2DPrTXaeb98=
 k8s.io/utils v0.0.0-20240310230437-4693a0247e57 h1:gbqbevonBh57eILzModw6mrkbwM0gQBEuevE/AaBsHY=
 k8s.io/utils v0.0.0-20240310230437-4693a0247e57/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd h1:EDPBXCAspyGV4jQlpZSudPeMmr1bNJefnuqLsRAsHZo=

pkg/detector/detector.go

@@ -6,6 +6,7 @@ package detector
 import (
 	"github.com/intelops/tarian-detector/pkg/err"
 	"github.com/intelops/tarian-detector/pkg/eventparser"
+	"github.com/intelops/tarian-detector/pkg/k8s"
 )
 
 var detectorErr = err.New("detector.detector")
@@ -32,6 +33,7 @@ type EventsDetector struct {
 	totalRecordsCount int                     // totalRecordsCount is the total count of records.
 	totalDetectors    int                     // totalDetectors is the total number of detectors.
 	probeRecordsCount map[string]int          // probeRecordsCount is a map of probe names to their respective counts
+	watcher           *k8s.PodWatcher         // watcher is a pointer to a PodWatcher instance
 }
 
 // NewEventsDetector creates a new EventsDetector instance
@@ -79,6 +81,11 @@ func (t *EventsDetector) GetProbeCount() map[string]int {
 	return t.probeRecordsCount
 }
 
+// SetPodWatcher sets the pod watcher.
+func (t *EventsDetector) SetPodWatcher(w *k8s.PodWatcher) {
+	t.watcher = w
+}
+
 // Start initiates the event detection process. It iterates over the map of each detector,
 // starts a goroutine for each map. These goroutines continuously read events from the maps
 // and send them to the event queue. If the detector is closed, the goroutines stop reading events.
@@ -112,6 +119,10 @@ func (t *EventsDetector) Start() error {
 
 	t.started = true
 
+	if t.watcher != nil {
+		t.watcher.Start()
+	}
+
 	return nil
 }
 
@@ -131,23 +142,20 @@ func (t *EventsDetector) Close() error {
 
 // ReadAsInterface reads a byte array from the event queue, parses it, and increments the total count.
 // It also checks for the presence of an event ID and increments the probe count if found.
-func (t *EventsDetector) ReadAsInterface() (map[string]any, error) {
+func (t *EventsDetector) ReadAsInterface() (eventparser.TarianDetectorEvent, error) {
 	eventparser.LoadTarianEvents()
 	r := <-t.eventQueue
 	if r.err != nil {
-		return map[string]any{}, detectorErr.Throwf("%v", r.err)
+		return eventparser.TarianDetectorEvent{}, detectorErr.Throwf("%v", r.err)
 	}
 
 	t.incrementTotalCount()
-	data, err := eventparser.ParseByteArray(r.eventData)
+	data, err := eventparser.ParseByteArray(t.watcher, r.eventData)
 	if err != nil {
 		return data, detectorErr.Throwf("%v", err)
 	}
 
-	probe, ok := data["eventId"]
-	if ok {
-		t.probeCount(probe.(string))
-	}
+	t.probeCount(data.EventId)
 
 	return data, nil
 }

pkg/eBPF/handler.go

@@ -12,9 +12,10 @@ var handlerErr = err.New("ebpf.handler")
 
 // Handler represents an eBPF handler. It includes the name of the handler, a list of map readers, and a list of probe links.
 type Handler struct {
-	name       string      // Name of the handler
-	mapReaders []any       // List of map readers
-	probeLinks []link.Link // List of probe links
+	name          string      // Name of the handler
+	mapReaders    []any       // List of map readers
+	probeLinks    []link.Link // List of probe links
+	countPrograms int
 }
 
 // NewHandler creates a new eBPF handler with the given name.
@@ -43,7 +44,7 @@ func (h *Handler) ReadAsInterface() ([]func() ([]byte, error), error) {
 
 // Count returns the number of probe links in the handler.
 func (h *Handler) Count() int {
-	return len(h.probeLinks)
+	return h.countPrograms
 }
 
 // Close detaches probes and closes map readers.

pkg/eBPF/handler_test.go

@@ -147,9 +147,10 @@ func TestHandler_Count(t *testing.T) {
 	l, _ := link.Kprobe("vprintk", prog, nil)
 
 	type fields struct {
-		name       string
-		mapReaders []any
-		probeLinks []link.Link
+		name          string
+		mapReaders    []any
+		probeLinks    []link.Link
+		countPrograms int
 	}
 	tests := []struct {
 		name   string
@@ -159,18 +160,20 @@ func TestHandler_Count(t *testing.T) {
 		{
 			name: "valid values",
 			fields: fields{
-				name:       "test",
-				mapReaders: make([]any, 0),
-				probeLinks: make([]link.Link, 0),
+				name:          "test",
+				mapReaders:    make([]any, 0),
+				probeLinks:    make([]link.Link, 0),
+				countPrograms: 0,
 			},
 			want: 0,
 		},
 		{
 			name: "add an item to the probelink",
 			fields: fields{
-				name:       "test",
-				mapReaders: nil,
-				probeLinks: []link.Link{l, l, l},
+				name:          "test",
+				mapReaders:    nil,
+				probeLinks:    []link.Link{l, l, l},
+				countPrograms: 3,
 			},
 			want: 3,
 		},
@@ -179,9 +182,10 @@ func TestHandler_Count(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			h := &Handler{
-				name:       tt.fields.name,
-				mapReaders: tt.fields.mapReaders,
-				probeLinks: tt.fields.probeLinks,
+				name:          tt.fields.name,
+				mapReaders:    tt.fields.mapReaders,
+				probeLinks:    tt.fields.probeLinks,
+				countPrograms: tt.fields.countPrograms,
 			}
 			if got := h.Count(); got != tt.want {
 				t.Errorf("Handler.Count() = %v, want %v", got, tt.want)