Add support for detecting AWS ALB ingress resources and auto-allowing traffic (#476)

Co-authored-by: omri.s <[email protected]>

Author: orishoshan

src/operator/controllers/external_traffic/network_policy.go

@@ -42,17 +42,23 @@ type NetworkPolicyHandler struct {
 	client client.Client
 	scheme *runtime.Scheme
 	injectablerecorder.InjectableRecorder
-	allowExternalTraffic        allowexternaltraffic.Enum
-	ingressControllerIdentities []serviceidentity.ServiceIdentity
+	allowExternalTraffic         allowexternaltraffic.Enum
+	ingressControllerIdentities  []serviceidentity.ServiceIdentity
+	ingressControllerALBAllowAll bool
 }
 
 func NewNetworkPolicyHandler(
 	client client.Client,
 	scheme *runtime.Scheme,
 	allowExternalTraffic allowexternaltraffic.Enum,
 	ingressControllerIdentities []serviceidentity.ServiceIdentity,
+	ingressControllerALBAllowAll bool,
 ) *NetworkPolicyHandler {
-	return &NetworkPolicyHandler{client: client, scheme: scheme, allowExternalTraffic: allowExternalTraffic, ingressControllerIdentities: ingressControllerIdentities}
+	return &NetworkPolicyHandler{client: client, scheme: scheme, allowExternalTraffic: allowExternalTraffic, ingressControllerIdentities: ingressControllerIdentities, ingressControllerALBAllowAll: ingressControllerALBAllowAll}
+}
+
+func (r *NetworkPolicyHandler) SetIngressControllerALBAllowAll(ingressControllerALBAllowAll bool) {
+	r.ingressControllerALBAllowAll = ingressControllerALBAllowAll
 }
 
 func (r *NetworkPolicyHandler) createOrUpdateNetworkPolicy(
@@ -129,7 +135,7 @@ func (r *NetworkPolicyHandler) buildNetworkPolicyObjectForEndpoints(
 
 	rule := v1.NetworkPolicyIngressRule{}
 	// Only limit netpol if there is an ingress controller restriction configured AND the service is not directly exposed.
-	if len(r.ingressControllerIdentities) != 0 && svc.Spec.Type == corev1.ServiceTypeClusterIP {
+	if len(r.ingressControllerIdentities) != 0 && svc.Spec.Type == corev1.ServiceTypeClusterIP && !(r.ingressControllerALBAllowAll && isIngressListHasInternetFacingAWSALB(ingressList.Items)) {
 		for _, ingressController := range r.ingressControllerIdentities {
 			rule.From = append(rule.From, v1.NetworkPolicyPeer{
 				PodSelector: &metav1.LabelSelector{

src/operator/controllers/external_traffic/network_policy_test.go

@@ -22,7 +22,7 @@ type NetworkPolicyHandlerTestSuite struct {
 
 func (s *NetworkPolicyHandlerTestSuite) SetupTest() {
 	s.MocksSuiteBase.SetupTest()
-	s.handler = NewNetworkPolicyHandler(s.Client, &runtime.Scheme{}, allowexternaltraffic.IfBlockedByOtterize, make([]serviceidentity.ServiceIdentity, 0))
+	s.handler = NewNetworkPolicyHandler(s.Client, &runtime.Scheme{}, allowexternaltraffic.IfBlockedByOtterize, make([]serviceidentity.ServiceIdentity, 0), false)
 }
 
 func (s *NetworkPolicyHandlerTestSuite) TestNetworkPolicyHandler_HandleBeforeAccessPolicyRemoval_createWhenNoIntentsEnabled_doNothing() {

src/operator/controllers/external_traffic/service_uploader.go

@@ -146,10 +146,12 @@ func convertToCloudExternalService(svc corev1.Service, identity serviceidentity.
 
 	// Remember to update the cache key that determines whether an update is needed.
 	serviceInput := graphqlclient.ExternallyAccessibleServiceInput{
-		Namespace:         identity.Namespace,
-		ServerName:        identity.Name,
-		ReferredByIngress: ReferredByIngress,
-		ServiceType:       cloudServiceType,
+		Namespace:                      identity.Namespace,
+		ServerName:                     identity.Name,
+		ReferredByIngress:              ReferredByIngress,
+		ServiceType:                    cloudServiceType,
+		ServiceName:                    svc.Name,
+		HasInternetFacingAWSALBIngress: isIngressListHasInternetFacingAWSALB(referringIngressList.Items),
 	}
 	return serviceInput, true, nil
 }

src/operator/controllers/external_traffic/service_uploader_test.go

@@ -227,18 +227,21 @@ func (s *ServiceUploaderTestSuite) TestUploadNamespaceServices() {
 		{
 			Namespace:         testNamespace,
 			ServerName:        podForServiceWithIngressName,
+			ServiceName:       serviceWithIngressName,
 			ReferredByIngress: true,
 			ServiceType:       graphqlclient.KubernetesServiceTypeClusterIp,
 		},
 		{
 			Namespace:         testNamespace,
 			ServerName:        podForServiceWithNodePortName,
+			ServiceName:       serviceWithNodePortName,
 			ReferredByIngress: false,
 			ServiceType:       graphqlclient.KubernetesServiceTypeNodePort,
 		},
 		{
 			Namespace:         testNamespace,
 			ServerName:        podForServiceWithLoadBalancerName,
+			ServiceName:       serviceWithLoadBalancerName,
 			ReferredByIngress: false,
 			ServiceType:       graphqlclient.KubernetesServiceTypeLoadBalancer,
 		},

src/operator/controllers/external_traffic/shared.go

@@ -1,6 +1,7 @@
 package external_traffic
 
 import (
+	"github.com/samber/lo"
 	"k8s.io/api/networking/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 )
@@ -23,3 +24,18 @@ func serviceNamesFromIngress(ingress *v1.Ingress) sets.Set[string] {
 
 	return serviceNames
 }
+
+func isIngressListHasInternetFacingAWSALB(ingressList []v1.Ingress) bool {
+	return lo.SomeBy(ingressList, func(ingress v1.Ingress) bool {
+		if ingress.Annotations == nil {
+			return false
+		}
+
+		scheme, ok := ingress.Annotations["alb.ingress.kubernetes.io/scheme"]
+		if !ok {
+			return false
+		}
+
+		return scheme == "internet-facing"
+	})
+}

src/operator/controllers/intents_reconcilers/external_traffic_network_policy/external_traffic_network_policy_test.go

@@ -92,7 +92,7 @@ func (s *ExternalNetworkPolicyReconcilerTestSuite) SetupTest() {
 	testName := s.T().Name()
 	isShadowMode := strings.Contains(testName, "ShadowMode")
 	defaultActive := !isShadowMode
-	netpolHandler := external_traffic.NewNetworkPolicyHandler(s.Mgr.GetClient(), s.TestEnv.Scheme, allowexternaltraffic.IfBlockedByOtterize, make([]serviceidentity.ServiceIdentity, 0))
+	netpolHandler := external_traffic.NewNetworkPolicyHandler(s.Mgr.GetClient(), s.TestEnv.Scheme, allowexternaltraffic.IfBlockedByOtterize, make([]serviceidentity.ServiceIdentity, 0), false)
 	s.defaultDenyReconciler = protected_service_reconcilers.NewDefaultDenyReconciler(s.Mgr.GetClient(), netpolHandler, true)
 	netpolReconciler := networkpolicy.NewReconciler(s.Mgr.GetClient(), s.TestEnv.Scheme, netpolHandler, []string{}, goset.NewSet[string](), true, defaultActive, []networkpolicy.IngressRuleBuilder{builders.NewIngressNetpolBuilder(), builders.NewPortNetworkPolicyReconciler(s.Mgr.GetClient())}, nil)
 	serviceIdResolver := serviceidresolver.NewResolver(s.Mgr.GetClient())
@@ -817,7 +817,7 @@ func (s *ExternalNetworkPolicyReconcilerTestSuite) TestEndpointsReconcilerNetwor
 
 	s.AddNodePortService(nodePortServiceName, podIps, podLabels)
 
-	netpolHandler := external_traffic.NewNetworkPolicyHandler(s.Mgr.GetClient(), s.TestEnv.Scheme, allowexternaltraffic.Off, make([]serviceidentity.ServiceIdentity, 0))
+	netpolHandler := external_traffic.NewNetworkPolicyHandler(s.Mgr.GetClient(), s.TestEnv.Scheme, allowexternaltraffic.Off, make([]serviceidentity.ServiceIdentity, 0), false)
 	endpointReconcilerWithEnforcementDisabled := external_traffic.NewEndpointsReconciler(s.Mgr.GetClient(), netpolHandler)
 	recorder := record.NewFakeRecorder(10)
 	endpointReconcilerWithEnforcementDisabled.InjectRecorder(recorder)

src/operator/controllers/intents_reconcilers/external_traffic_network_policy/external_traffic_network_policy_with_ingress_controllers_configured_test.go

@@ -58,6 +58,7 @@ type ExternalNetworkPolicyReconcilerWithIngressControllersConfiguredTestSuite st
 	EffectivePolicyIntentsReconciler *intents_reconcilers.ServiceEffectivePolicyIntentsReconciler
 	podWatcher                       *pod_reconcilers.PodWatcher
 	defaultDenyReconciler            *protected_service_reconcilers.DefaultDenyReconciler
+	netpolHandler                    *external_traffic.NetworkPolicyHandler
 }
 
 func (s *ExternalNetworkPolicyReconcilerWithIngressControllersConfiguredTestSuite) SetupSuite() {
@@ -105,7 +106,7 @@ func (s *ExternalNetworkPolicyReconcilerWithIngressControllersConfiguredTestSuit
 			Namespace: ingressControllerNamespace,
 			Name:      ingressControllerName,
 		},
-	})
+	}, false)
 	s.defaultDenyReconciler = protected_service_reconcilers.NewDefaultDenyReconciler(s.Mgr.GetClient(), netpolHandler, true)
 	netpolReconciler := networkpolicy.NewReconciler(s.Mgr.GetClient(), s.TestEnv.Scheme, netpolHandler, []string{}, goset.NewSet[string](), true, defaultActive, []networkpolicy.IngressRuleBuilder{builders.NewIngressNetpolBuilder(), builders.NewPortNetworkPolicyReconciler(s.Mgr.GetClient())}, nil)
 	serviceIdResolver := serviceidresolver.NewResolver(s.Mgr.GetClient())
@@ -123,6 +124,8 @@ func (s *ExternalNetworkPolicyReconcilerWithIngressControllersConfiguredTestSuit
 	s.IngressReconciler.InjectRecorder(recorder)
 	s.Require().NoError(err)
 
+	s.netpolHandler = netpolHandler
+
 	controller := gomock.NewController(s.T())
 	serviceEffectivePolicyReconciler := podreconcilersmocks.NewMockGroupReconciler(controller)
 	s.podWatcher = pod_reconcilers.NewPodWatcher(s.Mgr.GetClient(), recorder, []string{}, true, true, goset.NewSet[string](), &mocks.MockIntentsReconcilerForTestEnv{}, serviceEffectivePolicyReconciler)
@@ -899,7 +902,7 @@ func (s *ExternalNetworkPolicyReconcilerWithIngressControllersConfiguredTestSuit
 			Name:      ingressControllerName,
 			Kind:      "Deployment",
 		},
-	})
+	}, false)
 	endpointReconcilerWithEnforcementDisabled := external_traffic.NewEndpointsReconciler(s.Mgr.GetClient(), netpolHandler)
 	recorder := record.NewFakeRecorder(10)
 	endpointReconcilerWithEnforcementDisabled.InjectRecorder(recorder)
@@ -925,6 +928,145 @@ func (s *ExternalNetworkPolicyReconcilerWithIngressControllersConfiguredTestSuit
 	}
 }
 
+func (s *ExternalNetworkPolicyReconcilerWithIngressControllersConfiguredTestSuite) TestNetworkPolicyForAWSALBExemption_enabled() {
+	serviceName := "ingress-service"
+	ingressName := "test-ingress-alb"
+	ingressNamespace := s.TestNamespace
+	s.netpolHandler.SetIngressControllerALBAllowAll(true)
+
+	// Add Ingress with the annotation "alb.ingress.kubernetes.io/scheme": "internet-facing"
+	ingress := s.AddIngressWithAnnotation(ingressName, ingressNamespace, serviceName, map[string]string{
+		"alb.ingress.kubernetes.io/scheme": "internet-facing",
+	})
+
+	intents, err := s.AddIntents("test-intents", "test-client", "Deployment", []otterizev2alpha1.Target{{
+		Service: &otterizev2alpha1.ServiceTarget{Name: ingress.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Name},
+	},
+	})
+	s.Require().NoError(err)
+
+	_, err = s.EffectivePolicyIntentsReconciler.Reconcile(context.Background(), ctrl.Request{
+		NamespacedName: types.NamespacedName{
+			Namespace: intents.Namespace,
+			Name:      intents.Name,
+		},
+	})
+
+	s.Require().NoError(err)
+
+	// Reconcile the ingress
+	res, err := s.IngressReconciler.Reconcile(context.Background(), ctrl.Request{
+		NamespacedName: types.NamespacedName{Namespace: ingressNamespace, Name: ingressName},
+	})
+	s.Require().NoError(err)
+	s.Require().Empty(res)
+
+	// Verify that the network policy allows all ingress traffic
+	np := &v1.NetworkPolicy{}
+	policyName := fmt.Sprintf(external_traffic.OtterizeExternalNetworkPolicyNameTemplate, serviceName)
+	s.WaitUntilCondition(func(assert *assert.Assertions) {
+		err := s.Mgr.GetClient().Get(context.Background(), types.NamespacedName{Namespace: ingressNamespace, Name: policyName}, np)
+		assert.NoError(err)
+		assert.NotEmpty(np)
+		assert.Len(np.Spec.Ingress, 1)
+		if len(np.Spec.Ingress) == 1 {
+			assert.Len(np.Spec.Ingress[0].From, 0) // Allow all ingress traffic
+		}
+	})
+}
+
+func (s *ExternalNetworkPolicyReconcilerWithIngressControllersConfiguredTestSuite) TestNetworkPolicyForAWSALBExemption_disabled() {
+	serviceName := "ingress-service"
+	ingressName := "test-ingress-alb"
+	ingressNamespace := s.TestNamespace
+	s.netpolHandler.SetIngressControllerALBAllowAll(false)
+
+	// Add Ingress with the annotation "alb.ingress.kubernetes.io/scheme": "internet-facing"
+	ingress := s.AddIngressWithAnnotation(ingressName, ingressNamespace, serviceName, map[string]string{
+		"alb.ingress.kubernetes.io/scheme": "internet-facing",
+	})
+
+	intents, err := s.AddIntents("test-intents", "test-client", "Deployment", []otterizev2alpha1.Target{{
+		Service: &otterizev2alpha1.ServiceTarget{Name: ingress.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Name},
+	},
+	})
+	s.Require().NoError(err)
+
+	_, err = s.EffectivePolicyIntentsReconciler.Reconcile(context.Background(), ctrl.Request{
+		NamespacedName: types.NamespacedName{
+			Namespace: intents.Namespace,
+			Name:      intents.Name,
+		},
+	})
+
+	s.Require().NoError(err)
+
+	// Reconcile the ingress
+	res, err := s.IngressReconciler.Reconcile(context.Background(), ctrl.Request{
+		NamespacedName: types.NamespacedName{Namespace: ingressNamespace, Name: ingressName},
+	})
+	s.Require().NoError(err)
+	s.Require().Empty(res)
+
+	// Verify that the network policy allows all ingress traffic
+	np := &v1.NetworkPolicy{}
+	policyName := fmt.Sprintf(external_traffic.OtterizeExternalNetworkPolicyNameTemplate, serviceName)
+	s.WaitUntilCondition(func(assert *assert.Assertions) {
+		err := s.Mgr.GetClient().Get(context.Background(), types.NamespacedName{Namespace: ingressNamespace, Name: policyName}, np)
+		assert.NoError(err)
+		assert.NotEmpty(np)
+		assert.Len(np.Spec.Ingress, 1)
+		if len(np.Spec.Ingress) == 1 {
+			assert.Len(np.Spec.Ingress[0].From, 1) // Only allow traffic from the ingress controller
+		}
+	})
+}
+
+func (s *ExternalNetworkPolicyReconcilerWithIngressControllersConfiguredTestSuite) AddIngressWithAnnotation(name, namespace, serviceName string, annotations map[string]string) *v1.Ingress {
+	ingress := &v1.Ingress{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        name,
+			Namespace:   namespace,
+			Annotations: annotations,
+		},
+		Spec: v1.IngressSpec{
+			Rules: []v1.IngressRule{
+				{
+					Host: "example.com",
+					IngressRuleValue: v1.IngressRuleValue{
+						HTTP: &v1.HTTPIngressRuleValue{
+							Paths: []v1.HTTPIngressPath{
+								{
+									Path:     "/",
+									PathType: lo.ToPtr(v1.PathTypePrefix),
+									Backend: v1.IngressBackend{
+										Service: &v1.IngressServiceBackend{
+											Name: serviceName,
+											Port: v1.ServiceBackendPort{
+												Number: 80,
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	s.Require().NoError(s.Mgr.GetClient().Create(context.Background(), ingress))
+	s.WaitForObjectToBeCreated(ingress)
+
+	s.AddDeploymentWithService(serviceName, []string{"3.3.3.3"}, map[string]string{"app": "test"}, nil)
+
+	// the ingress reconciler expect the pod watcher labels in order to work
+	_, err := s.podWatcher.Reconcile(context.Background(), ctrl.Request{NamespacedName: types.NamespacedName{Namespace: s.TestNamespace, Name: serviceName + "-0"}})
+	s.Require().NoError(err)
+
+	return ingress
+}
+
 func TestExternalNetworkPolicyReconcilerWithIngressControllersConfiguredTestSuite(t *testing.T) {
 	suite.Run(t, new(ExternalNetworkPolicyReconcilerWithIngressControllersConfiguredTestSuite))
 }

src/operator/controllers/intents_reconcilers/external_traffic_network_policy/external_traffic_network_policy_with_no_intents_test.go

@@ -85,7 +85,7 @@ func (s *ExternalNetworkPolicyReconcilerWithNoIntentsTestSuite) SetupTest() {
 	s.Require().NoError((&otterizev2alpha1.ClientIntents{}).SetupWebhookWithManager(s.Mgr, intentsValidator2))
 
 	recorder := s.Mgr.GetEventRecorderFor("intents-operator")
-	netpolHandler := external_traffic.NewNetworkPolicyHandler(s.Mgr.GetClient(), s.TestEnv.Scheme, allowexternaltraffic.Always, make([]serviceidentity.ServiceIdentity, 0))
+	netpolHandler := external_traffic.NewNetworkPolicyHandler(s.Mgr.GetClient(), s.TestEnv.Scheme, allowexternaltraffic.Always, make([]serviceidentity.ServiceIdentity, 0), false)
 	netpolReconciler := networkpolicy.NewReconciler(s.Mgr.GetClient(), s.TestEnv.Scheme, netpolHandler, []string{}, goset.NewSet[string](), true, true, []networkpolicy.IngressRuleBuilder{builders.NewIngressNetpolBuilder()}, nil)
 	serviceIdResolver := serviceidresolver.NewResolver(s.Mgr.GetClient())
 	groupReconciler := effectivepolicy.NewGroupReconciler(s.Mgr.GetClient(), s.TestEnv.Scheme, serviceIdResolver, netpolReconciler)
@@ -246,7 +246,7 @@ func (s *ExternalNetworkPolicyReconcilerWithNoIntentsTestSuite) TestEndpointsRec
 
 	s.AddNodePortService(nodePortServiceName, podIps, podLabels)
 
-	netpolHandler := external_traffic.NewNetworkPolicyHandler(s.Mgr.GetClient(), s.TestEnv.Scheme, allowexternaltraffic.Off, make([]serviceidentity.ServiceIdentity, 0))
+	netpolHandler := external_traffic.NewNetworkPolicyHandler(s.Mgr.GetClient(), s.TestEnv.Scheme, allowexternaltraffic.Off, make([]serviceidentity.ServiceIdentity, 0), false)
 	endpointReconcilerWithEnforcementDisabled := external_traffic.NewEndpointsReconciler(s.Mgr.GetClient(), netpolHandler)
 	recorder := record.NewFakeRecorder(10)
 	endpointReconcilerWithEnforcementDisabled.InjectRecorder(recorder)

src/operator/main.go

@@ -197,7 +197,7 @@ func main() {
 
 	kafkaServersStore := kafkaacls.NewServersStore(tlsSource, enforcementConfig.EnableKafkaACL, kafkaacls.NewKafkaIntentsAdmin, enforcementConfig.EnforcementDefaultState)
 
-	extNetpolHandler := external_traffic.NewNetworkPolicyHandler(mgr.GetClient(), mgr.GetScheme(), allowExternalTraffic, operatorconfig.GetIngressControllerServiceIdentities())
+	extNetpolHandler := external_traffic.NewNetworkPolicyHandler(mgr.GetClient(), mgr.GetScheme(), allowExternalTraffic, operatorconfig.GetIngressControllerServiceIdentities(), viper.GetBool(operatorconfig.IngressControllerALBExemptKey))
 	endpointReconciler := external_traffic.NewEndpointsReconciler(mgr.GetClient(), extNetpolHandler)
 	ingressRulesBuilder := builders.NewIngressNetpolBuilder()
 

src/shared/operatorconfig/config.go

@@ -61,6 +61,8 @@ const (
 	TelemetryErrorsAPIKeyKey               = "telemetry-errors-api-key"
 	TelemetryErrorsAPIKeyDefault           = "60a78208a2b4fe714ef9fb3d3fdc0714"
 	AWSAccountsKey                         = "aws"
+	IngressControllerALBExemptKey          = "ingress-controllers-exempt-alb"
+	IngressControllerALBExemptDefault      = false
 	IngressControllerConfigKey             = "ingressControllers"
 )
 
@@ -76,6 +78,7 @@ func init() {
 	viper.SetDefault(AWSRolesAnywhereCertDirKey, AWSRolesAnywhereCertDirDefault)
 	viper.SetDefault(AWSRolesAnywherePrivKeyFilenameKey, AWSRolesAnywherePrivKeyFilenameDefault)
 	viper.SetDefault(AWSRolesAnywhereCertFilenameKey, AWSRolesAnywhereCertFilenameDefault)
+	viper.SetDefault(IngressControllerALBExemptKey, IngressControllerALBExemptDefault)
 	viper.SetDefault(KafkaServerTLSCertKey, "")
 	viper.SetDefault(KafkaServerTLSKeyKey, "")
 	viper.SetDefault(KafkaServerTLSCAKey, "")