[CWS-6786] Insertion of connect events in the activity tree

go.mod

@@ -41,7 +41,7 @@ require (
 	github.com/Azure/azure-sdk-for-go/sdk/security/keyvault/azsecrets v1.5.0
 	github.com/CycloneDX/cyclonedx-go v0.12.0
 	github.com/DATA-DOG/go-sqlmock v1.5.2
-	github.com/DataDog/agent-payload/v5 v5.0.210
+	github.com/DataDog/agent-payload/v5 v5.0.212-0.20260917144100-9eb352e40a01
 	github.com/DataDog/datadog-agent/comp/anomalydetection/observer/def v0.0.0-00010101000000-000000000000
 	github.com/DataDog/datadog-agent/comp/anomalydetection/recorder/def v0.0.0-00010101000000-000000000000
 	github.com/DataDog/datadog-agent/comp/anomalydetection/severityevents/def v0.0.0-00010101000000-000000000000

pkg/config/schema/yaml/system-probe-cws.yaml

@@ -716,6 +716,7 @@ properties:
             - open
             - dns
             - bind
+            - connect
             items:
               type: string
           excluded_images:

pkg/config/setup/system_probe_test.go

@@ -59,7 +59,7 @@ func TestSystemProbeDefaultConfig(t *testing.T) {
 		{key: "discovery.service_collection_min_process_age", defaultValue: time.Minute},
 		{key: "runtime_security_config.security_profile.v2.enabled", defaultValue: true},
 		{key: "runtime_security_config.security_profile.v2.max_dump_size", defaultValue: 2560},
-		{key: "runtime_security_config.security_profile.v2.event_types", defaultValue: []string{"exec", "open", "dns", "bind"}},
+		{key: "runtime_security_config.security_profile.v2.event_types", defaultValue: []string{"exec", "open", "dns", "bind", "connect"}},
 	} {
 		t.Run(tc.key, func(t *testing.T) {
 			switch expected := tc.defaultValue.(type) {

pkg/security/ebpf/c/include/hooks/network/connect.h

@@ -35,13 +35,15 @@ int __attribute__((always_inline)) sys_connect_ret_impl(void *ctx, int retval, e
         return 0;
     }
 
-    approve_syscall(syscall, connect_approvers);
-
-    // EAGAIN may be returned on Fedora 37 (kernel 6.0.7-301.fc37.x86_64)
+    // Bail out on failed connects before the approvers, otherwise a dropped event would still
+    // pollute connect_samples and suppress later successful connects to the same endpoint.
+    // EAGAIN may be returned on Fedora 37 (kernel 6.0.7-301.fc37.x86_64).
     if (IS_UNHANDLED_ERROR(retval) && retval != -EINPROGRESS && retval != -EAGAIN) {
         return 0;
     }
 
+    approve_syscall(syscall, connect_approvers);
+
     register_connecting_flow(syscall->connect.sk, syscall->connect.pid_tgid ? syscall->connect.pid_tgid : bpf_get_current_pid_tgid());
 
     // these probes are also loaded with the network probes, only send the event when a rule asks for it

pkg/security/security_profile/activity_tree/activity_tree.go

@@ -36,6 +36,7 @@ const (
 	eventTypeReason NodeDroppedReason = iota
 	invalidRootNodeReason
 	bindFamilyReason
+	connectFamilyReason
 	brokenEventReason
 
 	minNodeDroppedReason   = eventTypeReason
@@ -51,6 +52,8 @@ func (reason NodeDroppedReason) String() string {
 		return "invalid_root_node"
 	case bindFamilyReason:
 		return "bind_family"
+	case connectFamilyReason:
+		return "connect_family"
 	case brokenEventReason:
 		return "broken_event"
 	default:
@@ -68,6 +71,8 @@ func (reason NodeDroppedReason) Tag() string {
 		return "reason:invalid_root_node"
 	case bindFamilyReason:
 		return "reason:bind_family"
+	case connectFamilyReason:
+		return "reason:connect_family"
 	case brokenEventReason:
 		return "reason:broken_event"
 	default:
@@ -82,6 +87,8 @@ var (
 	ErrNotValidRootNode = errors.New("root node not valid")
 	// ErrInvalidBindFamily is returned when a bind event uses an unsupported address family
 	ErrInvalidBindFamily = errors.New("invalid bind address family")
+	// ErrInvalidConnectFamily is returned when a connect event uses an unsupported address family
+	ErrInvalidConnectFamily = errors.New("invalid connect address family")
 	// ErrIMDSMissingCredentials is returned when an IMDS response event has no access key ID
 	ErrIMDSMissingCredentials = errors.New("IMDS response without credentials")
 	// ErrIMDSMissingURL is returned when an IMDS request event has no URL
@@ -330,6 +337,10 @@ func (at *ActivityTree) ComputeActivityTreeStats() {
 		at.Stats.DNSNodes += int64(len(node.DNSNames))
 		at.Stats.SocketNodes += int64(len(node.Sockets))
 
+		for _, sock := range node.Sockets {
+			at.Stats.ConnectNodes += int64(len(sock.Connect))
+		}
+
 		for _, f := range node.Files {
 			fnodes = append(fnodes, f)
 		}
@@ -392,6 +403,7 @@ func IsExpectedFilterError(err error) bool {
 		errors.As(err, &pathResolutionNotCriticalErr) ||
 		errors.Is(err, ErrNotValidRootNode) ||
 		errors.Is(err, ErrInvalidBindFamily) ||
+		errors.Is(err, ErrInvalidConnectFamily) ||
 		errors.Is(err, ErrIMDSMissingCredentials) ||
 		errors.Is(err, ErrIMDSMissingURL)
 }
@@ -452,6 +464,14 @@ func (at *ActivityTree) isEventValid(event *model.Event, dryRun bool) (bool, err
 			}
 			return false, fmt.Errorf("%w: %s", ErrInvalidBindFamily, model.AddressFamily(event.Bind.AddrFamily))
 		}
+	case model.ConnectEventType:
+		// ignore non IPv4 / IPv6 connect events for now
+		if event.Connect.AddrFamily != unix.AF_INET && event.Connect.AddrFamily != unix.AF_INET6 {
+			if !dryRun {
+				at.Stats.counts[model.ConnectEventType].droppedCount[connectFamilyReason].Inc()
+			}
+			return false, fmt.Errorf("%w: %s", ErrInvalidConnectFamily, model.AddressFamily(event.Connect.AddrFamily))
+		}
 	case model.IMDSEventType:
 		// ignore IMDS answers without AccessKeyIDS
 		if event.IMDS.Type == model.IMDSResponseType && len(event.IMDS.AWS.SecurityCredentials.AccessKeyID) == 0 {
@@ -540,6 +560,9 @@ func (at *ActivityTree) insertEvent(event *model.Event, dryRun bool, insertMissi
 	case model.BindEventType:
 		newEntry, eventNodeBase := node.InsertBindEvent(event, imageTagID, generationType, at.Stats, dryRun)
 		return newEntry, node, eventNodeBase, nil
+	case model.ConnectEventType:
+		newEntry, eventNodeBase := node.InsertConnectEvent(event, imageTagID, generationType, at.Stats, dryRun)
+		return newEntry, node, eventNodeBase, nil
 	case model.SyscallsEventType:
 		return node.InsertSyscalls(event, imageTagID, at.SyscallsMask, at.Stats, dryRun), node, nil, nil
 	case model.NetworkFlowMonitorEventType:

pkg/security/security_profile/activity_tree/activity_tree_graph.go

@@ -422,7 +422,7 @@ func (at *ActivityTree) prepareSocketNode(n *SocketNode, data *utils.Graph, proc
 	for i, node := range n.Bind {
 		bindNode := &utils.Node{
 			ID:    processID.Derive(utils.NewNodeIDFromPtr(n), utils.NewNodeID(uint64(i+1))),
-			Label: "[" + node.IP + "]:" + strconv.FormatUint(uint64(node.Port), 10),
+			Label: "bind [" + node.IP + "]:" + strconv.FormatUint(uint64(node.Port), 10),
 			Size:  smallText,
 			Color: networkColor,
 			Shape: networkShape,
@@ -442,6 +442,31 @@ func (at *ActivityTree) prepareSocketNode(n *SocketNode, data *utils.Graph, proc
 		data.Nodes[bindNode.ID] = bindNode
 	}
 
+	// prepare connect nodes
+	bindCount := uint64(len(n.Bind))
+	for i, node := range n.Connect {
+		connectNode := &utils.Node{
+			ID:    processID.Derive(utils.NewNodeIDFromPtr(n), utils.NewNodeID(bindCount+uint64(i)+1)),
+			Label: "connect [" + node.IP + "]:" + strconv.FormatUint(uint64(node.Port), 10),
+			Size:  smallText,
+			Color: networkColor,
+			Shape: networkShape,
+		}
+
+		switch node.GenerationType {
+		case Runtime, Snapshot, Unknown:
+			connectNode.FillColor = networkRuntimeColor
+		case ProfileDrift:
+			connectNode.FillColor = networkProfileDriftColor
+		}
+		data.Edges = append(data.Edges, &utils.Edge{
+			From:  targetID,
+			To:    connectNode.ID,
+			Color: networkColor,
+		})
+		data.Nodes[connectNode.ID] = connectNode
+	}
+
 	return targetID
 }
 

pkg/security/security_profile/activity_tree/activity_tree_proto_dec_v1.go

@@ -496,6 +496,30 @@ func protoDecodeProtoSocket(sn *adproto.SocketNode, getIDFromImageTag func(strin
 		socketNode.Bind = append(socketNode.Bind, psn)
 	}
 
+	for _, connectNode := range sn.GetConnect() {
+		cn := &ConnectNode{
+			MatchedRules: make([]*model.MatchedRule, 0, len(connectNode.MatchedRules)),
+			Port:         uint16(connectNode.Port),
+			IP:           connectNode.Ip,
+			Protocol:     uint16(connectNode.Protocol),
+			NodeBase:     NewNodeBase(),
+		}
+
+		if connectNode.NodeBase != nil {
+			for tag, imageTagTimes := range connectNode.NodeBase.Seen {
+				firstSeen := ProtoDecodeTimestamp(imageTagTimes.FirstSeen)
+				lastSeen := ProtoDecodeTimestamp(imageTagTimes.LastSeen)
+				cn.RecordWithTimestamps(getIDFromImageTag(tag), firstSeen, lastSeen)
+			}
+		}
+
+		for _, rule := range connectNode.MatchedRules {
+			cn.MatchedRules = append(cn.MatchedRules, protoDecodeProtoMatchedRule(rule))
+		}
+
+		socketNode.Connect = append(socketNode.Connect, cn)
+	}
+
 	return socketNode
 }
 

pkg/security/security_profile/activity_tree/activity_tree_proto_enc_v1.go

@@ -387,8 +387,9 @@ func socketNodeToProto(sn *SocketNode, tagIDToImageTag func(id uint64) string) *
 	}
 
 	psn := &adproto.SocketNode{
-		Family: sn.Family,
-		Bind:   make([]*adproto.BindNode, 0, len(sn.Bind)),
+		Family:  sn.Family,
+		Bind:    make([]*adproto.BindNode, 0, len(sn.Bind)),
+		Connect: make([]*adproto.ConnectNode, 0, len(sn.Connect)),
 	}
 
 	for _, bn := range sn.Bind {
@@ -407,6 +408,22 @@ func socketNodeToProto(sn *SocketNode, tagIDToImageTag func(id uint64) string) *
 		psn.Bind = append(psn.Bind, pbn)
 	}
 
+	for _, cn := range sn.Connect {
+		pcn := &adproto.ConnectNode{
+			MatchedRules: make([]*adproto.MatchedRule, 0, len(cn.MatchedRules)),
+			Port:         uint32(cn.Port),
+			Ip:           cn.IP,
+			Protocol:     uint32(cn.Protocol),
+			NodeBase:     nodeBaseToProto(&cn.NodeBase, tagIDToImageTag),
+		}
+
+		for _, rule := range cn.MatchedRules {
+			pcn.MatchedRules = append(pcn.MatchedRules, matchedRuleToProto(rule))
+		}
+
+		psn.Connect = append(psn.Connect, pcn)
+	}
+
 	return psn
 }
 

pkg/security/security_profile/activity_tree/activity_tree_stats.go

@@ -28,6 +28,7 @@ type Stats struct {
 	IMDSNodes       int64
 	SyscallNodes    int64
 	FlowNodes       int64
+	ConnectNodes    int64
 	CapabilityNodes int64
 	SizeBytes       int64
 
@@ -78,6 +79,7 @@ func (stats *Stats) ApproximateSize() int64 {
 	total += stats.IMDSNodes * int64(unsafe.Sizeof(IMDSNode{}))
 	total += stats.SyscallNodes * int64(unsafe.Sizeof(SyscallNode{}))
 	total += stats.FlowNodes * int64(unsafe.Sizeof(FlowNode{}))
+	total += stats.ConnectNodes * int64(unsafe.Sizeof(ConnectNode{}))
 	total += stats.CapabilityNodes * int64(unsafe.Sizeof(CapabilityNode{}))
 	return total
 }

pkg/security/security_profile/activity_tree/process_node.go

@@ -16,6 +16,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"syscall"
 	"time"
 	"unsafe"
 
@@ -296,6 +297,14 @@ func (pn *ProcessNode) debug(w io.Writer, prefix string) {
 			fmt.Fprintf(w, "%s    - %s | %s\n", prefix, evt.CloudProvider, evt.Type)
 		}
 	}
+	for _, sock := range pn.Sockets {
+		if len(sock.Connect) > 0 {
+			fmt.Fprintf(w, "%s  connect (%s):\n", prefix, sock.Family)
+			for _, conn := range sock.Connect {
+				fmt.Fprintf(w, "%s    - %s:%d\n", prefix, conn.IP, conn.Port)
+			}
+		}
+	}
 	if len(pn.Children) > 0 {
 		fmt.Fprintf(w, "%s  children:\n", prefix)
 		for _, child := range pn.Children {
@@ -554,6 +563,41 @@ func (pn *ProcessNode) InsertBindEvent(evt *model.Event, imageTagID uint64, gene
 	return newNode, bindNodeBase
 }
 
+// InsertConnectEvent inserts a connect event in a process node. Returns whether a new entry was
+// added and the NodeBase of the matched or newly created ConnectNode.
+func (pn *ProcessNode) InsertConnectEvent(evt *model.Event, imageTagID uint64, generationType NodeGenerationType, stats *Stats, dryRun bool) (bool, *NodeBase) {
+	if evt.Connect.SyscallEvent.Retval != 0 &&
+		evt.Connect.SyscallEvent.Retval != -int64(syscall.EINPROGRESS) &&
+		evt.Connect.SyscallEvent.Retval != -int64(syscall.EAGAIN) {
+		return false, nil
+	}
+	var newNode bool
+	evtFamily := model.AddressFamily(evt.Connect.AddrFamily).String()
+
+	var sock *SocketNode
+	for _, s := range pn.Sockets {
+		if s.Family == evtFamily {
+			sock = s
+		}
+	}
+	if sock == nil {
+		sock = NewSocketNode(evtFamily, generationType)
+		if !dryRun {
+			stats.SocketNodes++
+			stats.SizeBytes += sock.size()
+			pn.Sockets = append(pn.Sockets, sock)
+		}
+		newNode = true
+	}
+
+	connectNew, connectNodeBase := sock.InsertConnectEvent(&evt.Connect, evt, imageTagID, generationType, evt.Rules, stats, dryRun)
+	if connectNew {
+		newNode = true
+	}
+
+	return newNode, connectNodeBase
+}
+
 // InsertCapabilitiesUsageEvent inserts a capabilities usage event in a process node
 func (pn *ProcessNode) InsertCapabilitiesUsageEvent(evt *model.Event, imageTagID uint64, stats *Stats, dryRun bool) bool {
 	hasNewCapabilitiesUsage := false
@@ -613,6 +657,12 @@ func (pn *ProcessNode) TagAllNodes(imageTagID uint64, timestamp time.Time) {
 	}
 	for _, sock := range pn.Sockets {
 		sock.AppendImageTagID(imageTagID, timestamp)
+		for _, bind := range sock.Bind {
+			bind.AppendImageTagID(imageTagID, timestamp)
+		}
+		for _, conn := range sock.Connect {
+			conn.AppendImageTagID(imageTagID, timestamp)
+		}
 	}
 	for _, scall := range pn.Syscalls {
 		scall.AppendImageTagID(imageTagID, timestamp)
@@ -816,16 +866,20 @@ func (pn *ProcessNode) EvictUnusedNodes(before time.Time, filepathsInProcessCach
 
 	// Note: NetworkDeviceNode doesn't embed NodeBase so we skip eviction for network devices
 
-	// Evict unused socket nodes
-	for i := len(pn.Sockets) - 1; i >= 0; i-- {
-		socketNode := pn.Sockets[i]
-		if socketNode.NodeBase.EvictBeforeTimestamp(before) > 0 {
-			if socketNode.SeenIsEmpty() {
-				removedBytes += socketNode.size()
-				pn.Sockets = append(pn.Sockets[:i], pn.Sockets[i+1:]...)
-			}
+	// Evict unused socket nodes: children age out by their own timestamps, and a socket is
+	// removed only once it holds no children (see SocketNode.evictBeforeTimestamp).
+	newSockets := pn.Sockets[:0]
+	for _, socketNode := range pn.Sockets {
+		socketEmpty, socketRemoved := socketNode.evictBeforeTimestamp(before)
+		removedBytes += socketRemoved
+		if socketEmpty {
+			removedBytes += socketNode.size()
+			continue
 		}
+		newSockets = append(newSockets, socketNode)
 	}
+	clear(pn.Sockets[len(newSockets):])
+	pn.Sockets = newSockets
 
 	// Evict unused capability nodes
 	for i := len(pn.Capabilities) - 1; i >= 0; i-- {

pkg/security/security_profile/activity_tree/size_test.go

@@ -147,6 +147,7 @@ func TestApproximateSize_LegacyShallowSemantics(t *testing.T) {
 		{"imds_only", Stats{IMDSNodes: 2}},
 		{"syscall_only", Stats{SyscallNodes: 4}},
 		{"flow_only", Stats{FlowNodes: 7}},
+		{"connect_only", Stats{ConnectNodes: 3}},
 		{"capability_only", Stats{CapabilityNodes: 6}},
 	}
 	for _, tt := range tests {