FIx bug where hostNetwork pods could be falsely detected as incoming traffic from outside of the cluster; Support resolution of LoadBalancer typed services using their node ports (#240)

omris94 · web-flow · commit c7bfa5a60804 · 2024-09-19T11:06:35.000+03:00
diff --git a/src/mapper/pkg/kubefinder/kubefinder.go b/src/mapper/pkg/kubefinder/kubefinder.go
@@ -21,15 +21,16 @@ import (
 )
 
 const (
-	podIPIndexField            = "ip"
-	endpointIPPortIndexField   = "ipPort"
-	serviceIPIndexField        = "spec.ip"
-	externalIPIndexField       = "spec.externalIPs"
-	portNumberIndexField       = "service.spec.ports.nodePort"
-	nodeIPIndexField           = "node.status.Addresses.ExternalIP"
-	IstioCanonicalNameLabelKey = "service.istio.io/canonical-name"
-	apiServerName              = "kubernetes"
-	apiServerNamespace         = "default"
+	podIPIndexField                     = "ip"
+	podIPIncludingHostNetworkIndexField = "ipAndHostNetwork"
+	endpointIPPortIndexField            = "ipPort"
+	serviceIPIndexField                 = "spec.ip"
+	externalIPIndexField                = "spec.externalIPs"
+	nodePortNumberIndexField            = "service.spec.ports.nodePort"
+	nodeIPIndexField                    = "node.status.Addresses.ExternalIP"
+	IstioCanonicalNameLabelKey          = "service.istio.io/canonical-name"
+	apiServerName                       = "kubernetes"
+	apiServerNamespace                  = "default"
 )
 
 type KubeFinder struct {
@@ -70,6 +71,23 @@ func (k *KubeFinder) initIndexes(ctx context.Context) error {
 		return errors.Wrap(err)
 	}
 
+	err = k.mgr.GetCache().IndexField(ctx, &corev1.Pod{}, podIPIncludingHostNetworkIndexField, func(object client.Object) []string {
+		res := make([]string, 0)
+		pod := object.(*corev1.Pod)
+
+		if pod.DeletionTimestamp != nil {
+			return res
+		}
+
+		for _, ip := range pod.Status.PodIPs {
+			res = append(res, ip.IP)
+		}
+		return res
+	})
+	if err != nil {
+		return errors.Wrap(err)
+	}
+
 	err = k.mgr.GetCache().IndexField(ctx, &corev1.Service{}, serviceIPIndexField, func(object client.Object) []string {
 		res := make([]string, 0)
 		svc := object.(*corev1.Service)
@@ -99,13 +117,15 @@ func (k *KubeFinder) initIndexes(ctx context.Context) error {
 		return errors.Wrap(err)
 	}
 
-	err = k.mgr.GetCache().IndexField(ctx, &corev1.Service{}, portNumberIndexField, func(object client.Object) []string {
+	err = k.mgr.GetCache().IndexField(ctx, &corev1.Service{}, nodePortNumberIndexField, func(object client.Object) []string {
+		// node ports are unique per service - so it can be used for indexing services
 		ports := sets.New[string]()
 		svc := object.(*corev1.Service)
 		if svc.DeletionTimestamp != nil {
 			return nil
 		}
-		if svc.Spec.Type != corev1.ServiceTypeNodePort {
+		// Only node port and load balancer typed services use node ports
+		if svc.Spec.Type != corev1.ServiceTypeNodePort && svc.Spec.Type != corev1.ServiceTypeLoadBalancer {
 			return nil
 		}
 
@@ -252,22 +272,22 @@ func (k *KubeFinder) ResolveIPToControlPlane(ctx context.Context, ip string) (*c
 }
 
 func (k *KubeFinder) ResolveIPToExternalAccessService(ctx context.Context, ip string, port int) (*corev1.Service, bool, error) {
-	nodePortService, ok, err := k.resolveNodePortService(ctx, ip, port)
+	nodePortService, ok, err := k.resolveServiceByNodeIPAndPort(ctx, ip, port)
 	if err != nil {
 		return nil, false, errors.Wrap(err)
 	}
 	if ok {
 		return nodePortService, true, nil
 	}
 
-	loadBalancerService, ok, err := k.resolveLoadBalancerService(ctx, ip, port)
+	loadBalancerService, ok, err := k.resolveLoadBalancerServiceByExternalIP(ctx, ip, port)
 	if err != nil {
 		return nil, false, errors.Wrap(err)
 	}
 	return loadBalancerService, ok, nil
 }
 
-func (k *KubeFinder) resolveLoadBalancerService(ctx context.Context, ip string, port int) (*corev1.Service, bool, error) {
+func (k *KubeFinder) resolveLoadBalancerServiceByExternalIP(ctx context.Context, ip string, port int) (*corev1.Service, bool, error) {
 	var services corev1.ServiceList
 	err := k.client.List(ctx, &services, client.MatchingFields{externalIPIndexField: ip})
 	if err != nil {
@@ -288,7 +308,7 @@ func (k *KubeFinder) resolveLoadBalancerService(ctx context.Context, ip string,
 	return &service, true, nil
 }
 
-func (k *KubeFinder) resolveNodePortService(ctx context.Context, ip string, port int) (*corev1.Service, bool, error) {
+func (k *KubeFinder) resolveServiceByNodeIPAndPort(ctx context.Context, ip string, port int) (*corev1.Service, bool, error) {
 	var nodes corev1.NodeList
 	err := k.client.List(ctx, &nodes, client.MatchingFields{nodeIPIndexField: ip})
 	if err != nil {
@@ -304,7 +324,7 @@ func (k *KubeFinder) resolveNodePortService(ctx context.Context, ip string, port
 
 	portString := fmt.Sprintf("%d", port)
 	var services corev1.ServiceList
-	err = k.client.List(ctx, &services, client.MatchingFields{portNumberIndexField: portString})
+	err = k.client.List(ctx, &services, client.MatchingFields{nodePortNumberIndexField: portString})
 	if err != nil {
 		return nil, false, errors.Wrap(err)
 	}
@@ -442,3 +462,51 @@ func (k *KubeFinder) ResolveOtterizeIdentityForService(ctx context.Context, svc
 	dstSvcIdentity.KubernetesService = lo.ToPtr(svc.Name)
 	return dstSvcIdentity, true, nil
 }
+
+func (k *KubeFinder) IsSrcIpClusterInternal(ctx context.Context, ip string) (bool, error) {
+	// Known issue: this function is currently missing support for services/endpoints, node.PodCIDR, and pods that were deleted.
+
+	isNode, err := k.IsNodeIP(ctx, ip)
+	if err != nil {
+		return false, errors.Wrap(err)
+	}
+	if isNode {
+		return true, nil
+	}
+
+	isPod, err := k.IsPodIp(ctx, ip)
+	if err != nil {
+		return false, errors.Wrap(err)
+	}
+	if isPod {
+		return true, nil
+	}
+
+	_, isControlPlane, err := k.ResolveIPToControlPlane(ctx, ip)
+	if err != nil {
+		return false, errors.Wrap(err)
+	}
+	if isControlPlane {
+		return true, nil
+	}
+
+	return false, nil
+}
+
+func (k *KubeFinder) IsPodIp(ctx context.Context, ip string) (bool, error) {
+	var pods corev1.PodList
+	err := k.client.List(ctx, &pods, client.MatchingFields{podIPIncludingHostNetworkIndexField: ip})
+	if err != nil {
+		return false, errors.Wrap(err)
+	}
+	return len(pods.Items) > 0, nil
+}
+
+func (k *KubeFinder) IsNodeIP(ctx context.Context, ip string) (bool, error) {
+	var nodes corev1.NodeList
+	err := k.client.List(ctx, &nodes, client.MatchingFields{nodeIPIndexField: ip})
+	if err != nil {
+		return false, errors.Wrap(err)
+	}
+	return len(nodes.Items) > 0, nil
+}
diff --git a/src/mapper/pkg/resolvers/resolver_test.go b/src/mapper/pkg/resolvers/resolver_test.go
@@ -1224,7 +1224,7 @@ func (s *ResolverTestSuite) TestResolveOtterizeIdentityIgnoreHostNetworkPods() {
 	podIP := "1.1.1.3"
 
 	pod3 := s.AddPodWithHostNetwork("pod3", podIP, map[string]string{"app": "test"}, nil, true)
-	s.AddService(serviceName, map[string]string{"app": "test"}, serviceIP, []*v1.Pod{pod3})
+	s.AddClusterIPService(serviceName, map[string]string{"app": "test"}, serviceIP, []*v1.Pod{pod3})
 	s.Require().True(s.Mgr.GetCache().WaitForCacheSync(context.Background()))
 
 	service := &v1.Service{}
@@ -1238,6 +1238,133 @@ func (s *ResolverTestSuite) TestResolveOtterizeIdentityIgnoreHostNetworkPods() {
 
 }
 
+func (s *ResolverTestSuite) TestTCPResultsFromHostNetworkPodsIgnored() {
+	// Add external service
+	internalServiceIp := "10.0.0.16"
+	externalServiceIP := "34.10.0.12"
+	servicePort := 9090
+	s.AddDeploymentWithIngressService("service1", []string{"1.1.1.1"}, map[string]string{"app": "service1"}, internalServiceIp, externalServiceIP, servicePort)
+
+	// Add host network pod
+	hostNetworkServiceName := "test-service"
+	hostNetworkServiceIP := "10.0.0.10"
+	hostNetworkPodIP := "1.1.1.3"
+	pod := s.AddPodWithHostNetwork("pod", hostNetworkPodIP, map[string]string{"app": "test"}, nil, true)
+	s.AddClusterIPService(hostNetworkServiceName, map[string]string{"app": "test"}, hostNetworkServiceIP, []*v1.Pod{pod})
+
+	s.Require().True(s.Mgr.GetCache().WaitForCacheSync(context.Background()))
+
+	// Report TCP results of traffic from host network pod to external service
+	packetTime := time.Now().Add(time.Minute)
+	_, err := test_gql_client.ReportTCPCaptureResults(context.Background(), s.client, test_gql_client.CaptureTCPResults{
+		Results: []test_gql_client.RecordedDestinationsForSrc{
+			{
+				SrcIp: hostNetworkPodIP,
+				Destinations: []test_gql_client.Destination{
+					{
+						Destination:     externalServiceIP,
+						DestinationIP:   nilable.From(externalServiceIP),
+						DestinationPort: nilable.From(servicePort),
+						LastSeen:        packetTime,
+					},
+				},
+			},
+		},
+	})
+	s.Require().NoError(err)
+
+	s.waitForCaptureResultsProcessed(10 * time.Second)
+
+	// Verify that the traffic from host network pod to external service is ignored
+	res, err := test_gql_client.ServiceIntents(context.Background(), s.client, nil)
+	s.Require().NoError(err)
+	s.Require().ElementsMatch(res.ServiceIntents, []test_gql_client.ServiceIntentsServiceIntents{})
+
+	intents := s.resolver.incomingTrafficHolder.GetNewIntentsSinceLastGet()
+	s.Require().Empty(intents)
+}
+
+func (s *ResolverTestSuite) TestTCPResultsFromExternalToPodSavedAsIncoming() {
+	// Create deployment
+	deploymentName := "coolz"
+	podIP := "1.1.1.3"
+	dep, _ := s.AddDeployment(deploymentName, []string{podIP}, map[string]string{"app": "coolz"})
+	s.Require().True(s.Mgr.GetCache().WaitForCacheSync(context.Background()))
+
+	// Report TCP results of traffic from external ip to pod
+	packetTime := time.Now().Add(time.Minute)
+	_, err := test_gql_client.ReportTCPCaptureResults(context.Background(), s.client, test_gql_client.CaptureTCPResults{
+		Results: []test_gql_client.RecordedDestinationsForSrc{
+			{
+				SrcIp: "8.8.8.8",
+				Destinations: []test_gql_client.Destination{
+					{
+						Destination:     podIP,
+						DestinationIP:   nilable.From(podIP),
+						DestinationPort: nilable.From(80),
+						LastSeen:        packetTime,
+					},
+				},
+			},
+		},
+	})
+	s.Require().NoError(err)
+
+	s.waitForCaptureResultsProcessed(10 * time.Second)
+
+	// Verify that the traffic from external ip to pod is saved as incoming
+	intents := s.resolver.incomingTrafficHolder.GetNewIntentsSinceLastGet()
+	s.Require().Len(intents, 1)
+	s.Require().Equal(dep.Name, intents[0].Intent.Server.Name)
+	s.Require().Equal(dep.Namespace, intents[0].Intent.Server.Namespace)
+	s.Require().Equal("8.8.8.8", intents[0].Intent.IP)
+}
+
+func (s *ResolverTestSuite) TestTCPResultsFromExternalToLoadBalancerServiceUsingNodeIpAndPortSavedAsIncoming() {
+	// Create deployment
+	deploymentName := "coolz"
+	podIP := "1.1.1.3"
+	serviceIP := "10.0.0.10"
+	dep, pods := s.AddDeployment(deploymentName, []string{podIP}, map[string]string{"app": "coolz"})
+	s.Require().True(s.Mgr.GetCache().WaitForCacheSync(context.Background()))
+	svc := s.AddLoadBalancerService("coolz", map[string]string{"app": "coolz"}, serviceIP, pods)
+	s.Require().True(s.Mgr.GetCache().WaitForCacheSync(context.Background()))
+	nodePort := svc.Spec.Ports[0].NodePort
+
+	nodes := v1.NodeList{}
+	err := s.Mgr.GetClient().List(context.Background(), &nodes)
+	s.Require().NoError(err)
+	s.Require().NotEmpty(nodes.Items)
+	nodeIP := nodes.Items[0].Status.Addresses[0].Address
+
+	// Report TCP results of traffic from external ip to nodeIp:nodePort
+	packetTime := time.Now().Add(time.Minute)
+	_, err = test_gql_client.ReportTCPCaptureResults(context.Background(), s.client, test_gql_client.CaptureTCPResults{
+		Results: []test_gql_client.RecordedDestinationsForSrc{
+			{
+				SrcIp: "8.8.8.8",
+				Destinations: []test_gql_client.Destination{
+					{
+						Destination:     nodeIP,
+						DestinationIP:   nilable.From(nodeIP),
+						DestinationPort: nilable.From(int(nodePort)),
+						LastSeen:        packetTime,
+					},
+				},
+			},
+		},
+	})
+	s.Require().NoError(err)
+	s.waitForCaptureResultsProcessed(10 * time.Second)
+
+	// Verify that the traffic from external ip to nodeIp:nodePort is saved as incoming
+	intents := s.resolver.incomingTrafficHolder.GetNewIntentsSinceLastGet()
+	s.Require().Len(intents, 1)
+	s.Require().Equal(dep.Name, intents[0].Intent.Server.Name)
+	s.Require().Equal(dep.Namespace, intents[0].Intent.Server.Namespace)
+	s.Require().Equal("8.8.8.8", intents[0].Intent.IP)
+}
+
 func TestRunSuite(t *testing.T) {
 	suite.Run(t, new(ResolverTestSuite))
 }
diff --git a/src/mapper/pkg/resolvers/schema.helpers.resolvers.go b/src/mapper/pkg/resolvers/schema.helpers.resolvers.go
@@ -376,31 +376,40 @@ func (r *Resolver) handleReportTCPCaptureResults(ctx context.Context, results mo
 	}
 
 	for _, captureItem := range results.Results {
-		logrus.Debugf("Handling TCP capture result from %s to %s:%d", captureItem.SrcIP, captureItem.Destinations[0].Destination, lo.FromPtr(captureItem.Destinations[0].DestinationPort))
-
-		srcSvcIdentity, err := r.discoverInternalSrcIdentity(ctx, captureItem)
-		if errors.Is(err, kubefinder.ErrNoPodFound) {
-			err := r.reportIncomingInternetTraffic(ctx, captureItem.SrcIP, captureItem.Destinations)
-			if err != nil {
-				logrus.WithError(err).Error("could not report incoming internet traffic")
-				continue
-			}
-		}
-
+		err := r.handleTCPCaptureResult(ctx, captureItem)
 		if err != nil {
-			logrus.WithError(err).Debugf("could not discover src identity for '%s'", captureItem.SrcIP)
-			continue
-		}
-
-		for _, dest := range captureItem.Destinations {
-			r.handleExternalIncomingTrafficTCPResult(ctx, srcSvcIdentity, dest)
+			logrus.WithError(err).
+				WithField("srcIp", captureItem.SrcIP).
+				WithField("srcHostname", captureItem.SrcHostname).
+				Error("could not handle TCP capture result")
 		}
 	}
 	telemetrysender.SendNetworkMapper(telemetriesgql.EventTypeIntentsDiscoveredCapture, len(results.Results))
 	r.gotResultsSignal()
 	return nil
 }
 
+func (r *Resolver) handleTCPCaptureResult(ctx context.Context, captureItem model.RecordedDestinationsForSrc) error {
+	logrus.Debugf("Handling TCP capture result from %s to %s:%d", captureItem.SrcIP, captureItem.Destinations[0].Destination, lo.FromPtr(captureItem.Destinations[0].DestinationPort))
+	isSrcInCluster, err := r.kubeFinder.IsSrcIpClusterInternal(ctx, captureItem.SrcIP)
+	if err != nil {
+		return errors.Wrap(err)
+	}
+	if !isSrcInCluster {
+		return errors.Wrap(r.reportIncomingInternetTraffic(ctx, captureItem.SrcIP, captureItem.Destinations))
+	}
+
+	srcSvcIdentity, err := r.discoverInternalSrcIdentity(ctx, captureItem)
+	if err != nil {
+		logrus.WithError(err).Debugf("could not discover src identity for '%s'", captureItem.SrcIP)
+		return nil
+	}
+	for _, dest := range captureItem.Destinations {
+		r.handleInternalTrafficTCPResult(ctx, srcSvcIdentity, dest)
+	}
+	return nil
+}
+
 func (r *Resolver) reportIncomingInternetTraffic(ctx context.Context, srcIP string, destinations []model.Destination) error {
 	for _, dest := range destinations {
 		destSvcIdentity, ok, err := r.resolveOtterizeIdentityForExternalAccessDestination(ctx, dest)
@@ -423,7 +432,7 @@ func (r *Resolver) reportIncomingInternetTraffic(ctx context.Context, srcIP stri
 	return nil
 }
 
-func (r *Resolver) handleExternalIncomingTrafficTCPResult(ctx context.Context, srcIdentity model.OtterizeServiceIdentity, dest model.Destination) {
+func (r *Resolver) handleInternalTrafficTCPResult(ctx context.Context, srcIdentity model.OtterizeServiceIdentity, dest model.Destination) {
 	lastSeen := dest.LastSeen
 	destIdentity, ok, err := r.resolveDestIdentity(ctx, dest, lastSeen)
 	if err != nil {
diff --git a/src/shared/testbase/testsuitebase.go b/src/shared/testbase/testsuitebase.go
