feat/fix(endpoints) : Ensure endpoints for a serivce honors traffic p…

…olicy

This PR ensures that the endpoints built for a service in EDS honor
SMI traffic target policies. Only those destinations endpoints are
programmed on the envoy, if its destination pod has a service account
specified as a destination in any of of the applicable traffic targets.

Resolves issue openservicemesh#1658

Signed-off-by: Sneha Chhabria <snchh@microsoft.com>

go.sum

@@ -1342,6 +1342,8 @@ k8s.io/client-go v0.18.6 h1:I+oWqJbibLSGsZj8Xs8F0aWVXJVIoUHWaaJV3kUN/Zw=
 k8s.io/client-go v0.18.6/go.mod h1:/fwtGLjYMS1MaM5oi+eXhKwG+1UHidUEXRh6cNsdO0Q=
 k8s.io/client-go v0.18.8 h1:SdbLpIxk5j5YbFr1b7fq8S7mDgDjYmUxSbszyoesoDM=
 k8s.io/client-go v0.18.8/go.mod h1:HqFqMllQ5NnQJNwjro9k5zMyfhZlOwpuTLVrxjkYSxU=
+k8s.io/client-go v1.5.1 h1:XaX/lo2/u3/pmFau8HN+sB5C/b4dc4Dmm2eXjBH4p1E=
+k8s.io/client-go v11.0.0+incompatible h1:LBbX2+lOwY9flffWlJM7f1Ct8V2SRNiMRDFeiwnJo9o=
 k8s.io/code-generator v0.18.0/go.mod h1:+UHX5rSbxmR8kzS+FAv7um6dtYrZokQvjHpDSYRVkTc=
 k8s.io/code-generator v0.18.5/go.mod h1:TgNEVx9hCyPGpdtCWA34olQYLkh3ok9ar7XfSsr8b6c=
 k8s.io/code-generator v0.18.6/go.mod h1:TgNEVx9hCyPGpdtCWA34olQYLkh3ok9ar7XfSsr8b6c=

pkg/catalog/endpoint.go

@@ -36,3 +36,45 @@ func (mc *MeshCatalog) GetResolvableServiceEndpoints(svc service.MeshService) ([
 	}
 	return endpoints, nil
 }
+
+// ListAllowedEndpointsForService returns only those endpoints for a service that belong to the allowed outbound service accounts
+// for the given downstream identity
+func (mc *MeshCatalog) ListAllowedEndpointsForService(downstreamIdentity service.K8sServiceAccount, upstreamSvc service.MeshService) ([]endpoint.Endpoint, error) {
+	var outboundEndpoints []endpoint.Endpoint
+	outboundEndpoints, err := mc.ListEndpointsForService(upstreamSvc)
+	log.Trace().Msgf("Outbound endpoints for service %s with downstream identity %s: %v", upstreamSvc, downstreamIdentity, outboundEndpoints)
+
+	destSvcAccounts, err := mc.ListAllowedOutboundServiceAccounts(downstreamIdentity)
+	if err != nil {
+		log.Error().Err(err).Msgf("Error looking outbound service accounts for proxy with identity %s", downstreamIdentity)
+		return nil, err
+	}
+
+	var allowedEndpoints []endpoint.Endpoint
+	for _, destSvcAccount := range destSvcAccounts {
+		podEndpoints := mc.listEndpointsforIdentity(destSvcAccount)
+		for _, ep := range outboundEndpoints {
+			for _, podIP := range podEndpoints {
+				if ep.IP.Equal(podIP.IP) {
+					allowedEndpoints = append(allowedEndpoints, ep)
+				}
+			}
+		}
+	}
+	log.Trace().Msgf("Allowed endpoints for service %s with downstream identity %s: %v", upstreamSvc, downstreamIdentity, allowedEndpoints)
+	return allowedEndpoints, nil
+}
+
+// listEndpointsforIdentity retrieves the list of IP addresses for the given service account
+func (mc *MeshCatalog) listEndpointsforIdentity(sa service.K8sServiceAccount) []endpoint.Endpoint {
+	var endpoints []endpoint.Endpoint
+	for _, provider := range mc.endpointsProviders {
+		ep := provider.ListEndpointsForIdentity(sa)
+		if len(ep) == 0 {
+			log.Trace().Msgf("[%s] No endpoints found for service account=%s", provider.GetID(), sa)
+			continue
+		}
+		endpoints = append(endpoints, ep...)
+	}
+	return endpoints
+}

pkg/catalog/endpoint_test.go

@@ -1,10 +1,27 @@
 package catalog
 
 import (
+	"context"
+	"net"
+	"testing"
+
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
+	"github.com/golang/mock/gomock"
+	"github.com/google/uuid"
+	access "github.com/servicemeshinterface/smi-sdk-go/pkg/apis/access/v1alpha3"
+	tassert "github.com/stretchr/testify/assert"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	testclient "k8s.io/client-go/kubernetes/fake"
+
+	"github.com/openservicemesh/osm/pkg/configurator"
+	"github.com/openservicemesh/osm/pkg/constants"
 	"github.com/openservicemesh/osm/pkg/endpoint"
+	k8s "github.com/openservicemesh/osm/pkg/kubernetes"
+	"github.com/openservicemesh/osm/pkg/service"
+	"github.com/openservicemesh/osm/pkg/smi"
 	"github.com/openservicemesh/osm/pkg/tests"
 )
 
@@ -35,3 +52,149 @@ var _ = Describe("Test catalog functions", func() {
 	})
 
 })
+
+func TestListAllowedEndpointsForService(t *testing.T) {
+	assert := tassert.New(t)
+
+	testCases := []struct {
+		name                     string
+		proxyIdentity            service.K8sServiceAccount
+		upstreamSvc              service.MeshService
+		trafficTargets           []*access.TrafficTarget
+		services                 []service.MeshService
+		outboundServices         map[service.K8sServiceAccount][]service.MeshService
+		outboundServiceEndpoints map[service.MeshService][]endpoint.Endpoint
+		expectedEndpoints        []endpoint.Endpoint
+	}{
+		{
+			name: `Traffic target defined for bookstore ServiceAccount.
+			This service account has only bookstore-v1 service on it.
+			Hence endpoints returned for bookstore-v1`,
+			proxyIdentity:  tests.BookbuyerServiceAccount,
+			upstreamSvc:    tests.BookstoreV1Service,
+			trafficTargets: []*access.TrafficTarget{&tests.TrafficTarget},
+			services:       []service.MeshService{tests.BookstoreV1Service},
+			outboundServices: map[service.K8sServiceAccount][]service.MeshService{
+				tests.BookstoreServiceAccount: {tests.BookstoreV1Service},
+			},
+			outboundServiceEndpoints: map[service.MeshService][]endpoint.Endpoint{
+				tests.BookstoreV1Service: {tests.Endpoint},
+			},
+			expectedEndpoints: []endpoint.Endpoint{tests.Endpoint},
+		},
+		{
+			name: `Traffic target defined for bookstore ServiceAccount. 
+			This service account has bookstore-v1 bookstore-v2 services,
+			but bookstore-v2 pod has service account bookstore-v2.
+			Hence no endpoints returned for bookstore-v2`,
+			proxyIdentity:  tests.BookbuyerServiceAccount,
+			upstreamSvc:    tests.BookstoreV2Service,
+			trafficTargets: []*access.TrafficTarget{&tests.TrafficTarget},
+			services:       []service.MeshService{tests.BookstoreV1Service, tests.BookstoreV2Service},
+			outboundServices: map[service.K8sServiceAccount][]service.MeshService{
+				tests.BookstoreServiceAccount: {tests.BookstoreV1Service, tests.BookstoreV2Service},
+			},
+			outboundServiceEndpoints: map[service.MeshService][]endpoint.Endpoint{
+				tests.BookstoreV1Service: {tests.Endpoint},
+				tests.BookstoreV2Service: {endpoint.Endpoint{
+					IP:   net.ParseIP("9.9.9.9"),
+					Port: endpoint.Port(tests.ServicePort),
+				}},
+			},
+			expectedEndpoints: []endpoint.Endpoint{},
+		},
+		{
+			name: `Traffic target defined for bookstore ServiceAccount. 
+			This service account has bookstore-v1 bookstore-v2 services,
+			since bookstore-v2 pod has service account bookstore-v2 which is allowed in the traffic target.
+			Hence endpoints returned for bookstore-v2`,
+			proxyIdentity:  tests.BookbuyerServiceAccount,
+			upstreamSvc:    tests.BookstoreV2Service,
+			trafficTargets: []*access.TrafficTarget{&tests.TrafficTarget, &tests.BookstoreV2TrafficTarget},
+			services:       []service.MeshService{tests.BookstoreV1Service, tests.BookstoreV2Service},
+			outboundServices: map[service.K8sServiceAccount][]service.MeshService{
+				tests.BookstoreServiceAccount:   {tests.BookstoreV1Service},
+				tests.BookstoreV2ServiceAccount: {tests.BookstoreV2Service},
+			},
+			outboundServiceEndpoints: map[service.MeshService][]endpoint.Endpoint{
+				tests.BookstoreV1Service: {tests.Endpoint},
+				tests.BookstoreV2Service: {endpoint.Endpoint{
+					IP:   net.ParseIP("9.9.9.9"),
+					Port: endpoint.Port(tests.ServicePort),
+				}},
+			},
+			expectedEndpoints: []endpoint.Endpoint{{
+				IP:   net.ParseIP("9.9.9.9"),
+				Port: endpoint.Port(tests.ServicePort),
+			}},
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			mockCtrl := gomock.NewController(t)
+			kubeClient := testclient.NewSimpleClientset()
+			defer mockCtrl.Finish()
+
+			mockConfigurator := configurator.NewMockConfigurator(mockCtrl)
+			mockKubeController := k8s.NewMockController(mockCtrl)
+			mockEndpointProvider := endpoint.NewMockProvider(mockCtrl)
+			mockMeshSpec := smi.NewMockMeshSpec(mockCtrl)
+
+			mc := MeshCatalog{
+				kubeController:     mockKubeController,
+				meshSpec:           mockMeshSpec,
+				endpointsProviders: []endpoint.Provider{mockEndpointProvider},
+			}
+
+			mockConfigurator.EXPECT().IsPermissiveTrafficPolicyMode().Return(false).AnyTimes()
+			mockMeshSpec.EXPECT().ListTrafficTargets().Return(tc.trafficTargets).AnyTimes()
+
+			mockEndpointProvider.EXPECT().GetID().Return("fake").AnyTimes()
+
+			for sa, services := range tc.outboundServices {
+				for _, svc := range services {
+					k8sService := tests.NewServiceFixture(svc.Name, svc.Namespace, map[string]string{})
+					mockKubeController.EXPECT().GetService(svc).Return(k8sService).AnyTimes()
+				}
+				mockEndpointProvider.EXPECT().GetServicesForServiceAccount(sa).Return(services, nil).AnyTimes()
+			}
+
+			for svc, endpoints := range tc.outboundServiceEndpoints {
+				mockEndpointProvider.EXPECT().ListEndpointsForService(svc).Return(endpoints).AnyTimes()
+			}
+
+			var pods []*v1.Pod
+			for sa, services := range tc.outboundServices {
+				for _, svc := range services {
+					podlabels := map[string]string{
+						tests.SelectorKey:                tests.SelectorValue,
+						constants.EnvoyUniqueIDLabelName: uuid.New().String(),
+					}
+					pod := tests.NewPodFixture(tests.Namespace, svc.Name, sa.Name, podlabels)
+					podEndpoints := tc.outboundServiceEndpoints[svc]
+					var podIps []v1.PodIP
+					for _, ep := range podEndpoints {
+						podIps = append(podIps, v1.PodIP{IP: ep.IP.String()})
+					}
+					pod.Status.PodIPs = podIps
+					pod.Spec.ServiceAccountName = sa.Name
+					_, err := kubeClient.CoreV1().Pods(tests.Namespace).Create(context.TODO(), &pod, metav1.CreateOptions{})
+					assert.Nil(err)
+					pods = append(pods, &pod)
+				}
+			}
+			mockKubeController.EXPECT().ListPods().Return(pods).AnyTimes()
+
+			for sa, services := range tc.outboundServices {
+				for _, svc := range services {
+					podEndpoints := tc.outboundServiceEndpoints[svc]
+					mockEndpointProvider.EXPECT().ListEndpointsForIdentity(sa).Return(podEndpoints).AnyTimes()
+				}
+			}
+
+			actual, err := mc.ListAllowedEndpointsForService(tc.proxyIdentity, tc.upstreamSvc)
+			assert.Nil(err)
+			assert.ElementsMatch(actual, tc.expectedEndpoints)
+		})
+	}
+}

pkg/catalog/fake.go

@@ -104,7 +104,7 @@ func NewFakeMeshCatalog(kubeClient kubernetes.Interface) *MeshCatalog {
 	mockKubeController.EXPECT().IsMonitoredNamespace(tests.BookbuyerService.Namespace).Return(true).AnyTimes()
 	mockKubeController.EXPECT().IsMonitoredNamespace(tests.BookwarehouseService.Namespace).Return(true).AnyTimes()
 	mockKubeController.EXPECT().ListServiceAccountsForService(tests.BookstoreV1Service).Return([]service.K8sServiceAccount{tests.BookstoreServiceAccount}, nil).AnyTimes()
-	mockKubeController.EXPECT().ListServiceAccountsForService(tests.BookstoreV2Service).Return([]service.K8sServiceAccount{tests.BookstoreServiceAccount}, nil).AnyTimes()
+	mockKubeController.EXPECT().ListServiceAccountsForService(tests.BookstoreV2Service).Return([]service.K8sServiceAccount{tests.BookstoreV2ServiceAccount}, nil).AnyTimes()
 	mockKubeController.EXPECT().ListServiceAccountsForService(tests.BookbuyerService).Return([]service.K8sServiceAccount{tests.BookbuyerServiceAccount}, nil).AnyTimes()
 
 	return NewMeshCatalog(mockKubeController, kubeClient, meshSpec, certManager,

pkg/catalog/helpers_test.go

@@ -51,7 +51,7 @@ func newFakeMeshCatalogForRoutes(t *testing.T, testParams testParams) *MeshCatal
 	}
 
 	// Create a bookstoreV2 pod
-	bookstoreV2Pod := tests.NewPodFixture(tests.BookstoreV2Service.Namespace, tests.BookstoreV2Service.Name, tests.BookstoreServiceAccountName, tests.PodLabels)
+	bookstoreV2Pod := tests.NewPodFixture(tests.BookstoreV2Service.Namespace, tests.BookstoreV2Service.Name, tests.BookstoreV2ServiceAccountName, tests.PodLabels)
 	if _, err := kubeClient.CoreV1().Pods(tests.BookstoreV2Service.Namespace).Create(context.TODO(), &bookstoreV2Pod, metav1.CreateOptions{}); err != nil {
 		t.Fatalf("Error creating new pod: %s", err.Error())
 	}

pkg/catalog/ingress_test.go

@@ -70,8 +70,14 @@ func newFakeMeshCatalog() *MeshCatalog {
 
 	certManager := tresor.NewFakeCertManager(cfg)
 
-	// Create a pod
-	pod := tests.NewPodFixture(tests.Namespace, "pod-name", tests.BookstoreServiceAccountName, tests.PodLabels)
+	// Create a Bookstore-v1 pod
+	pod := tests.NewPodFixture(tests.Namespace, tests.BookstoreV1Service.Name, tests.BookstoreServiceAccountName, tests.PodLabels)
+	if _, err := kubeClient.CoreV1().Pods(tests.Namespace).Create(context.TODO(), &pod, metav1.CreateOptions{}); err != nil {
+		GinkgoT().Fatalf("Error creating new fake Mesh Catalog: %s", err.Error())
+	}
+
+	// Create a Bookstore-v2 pod
+	pod = tests.NewPodFixture(tests.Namespace, tests.BookstoreV2Service.Name, tests.BookstoreV2ServiceAccountName, tests.PodLabels)
 	if _, err := kubeClient.CoreV1().Pods(tests.Namespace).Create(context.TODO(), &pod, metav1.CreateOptions{}); err != nil {
 		GinkgoT().Fatalf("Error creating new fake Mesh Catalog: %s", err.Error())
 	}

pkg/catalog/routes_test.go

@@ -558,6 +558,10 @@ func TestListTrafficPolicies(t *testing.T) {
 			input:  tests.BookstoreV1Service,
 			output: []trafficpolicy.TrafficTarget{tests.BookstoreV1TrafficPolicy},
 		},
+		{
+			input:  tests.BookstoreV2Service,
+			output: []trafficpolicy.TrafficTarget{tests.BookstoreV2TrafficPolicy},
+		},
 		{
 			input:  tests.BookbuyerService,
 			output: []trafficpolicy.TrafficTarget{tests.BookstoreV1TrafficPolicy, tests.BookstoreV2TrafficPolicy, tests.BookstoreApexTrafficPolicy},
@@ -592,7 +596,7 @@ func TestGetTrafficPoliciesForService(t *testing.T) {
 					HTTPRouteMatches: tests.BookstoreV1TrafficPolicy.HTTPRouteMatches,
 				},
 				{
-					Name:             utils.GetTrafficTargetName(tests.TrafficTargetName, tests.BookbuyerService, tests.BookstoreV2Service),
+					Name:             utils.GetTrafficTargetName(tests.BookstoreV2TrafficTargetName, tests.BookbuyerService, tests.BookstoreV2Service),
 					Destination:      tests.BookstoreV2Service,
 					Source:           tests.BookbuyerService,
 					HTTPRouteMatches: tests.BookstoreV2TrafficPolicy.HTTPRouteMatches,

pkg/catalog/types.go

@@ -89,6 +89,9 @@ type MeshCataloger interface {
 	// ListEndpointsForService returns the list of individual instance endpoint backing a service
 	ListEndpointsForService(service.MeshService) ([]endpoint.Endpoint, error)
 
+	// ListAllowedEndpointsForService returns the list of endpoints backing a service and its allowed service accounts
+	ListAllowedEndpointsForService(service.K8sServiceAccount, service.MeshService) ([]endpoint.Endpoint, error)
+
 	// GetResolvableServiceEndpoints returns the resolvable set of endpoint over which a service is accessible using its FQDN.
 	// These are the endpoint destinations we'd expect client applications sends the traffic towards to, when attempting to
 	// reach a specific service.

pkg/debugger/policy_test.go

@@ -53,6 +53,6 @@ func TestGetSMIPolicies(t *testing.T) {
 	responseRecorder := httptest.NewRecorder()
 	smiPoliciesHandler.ServeHTTP(responseRecorder, nil)
 	actualResponseBody := responseRecorder.Body.String()
-	expectedResponseBody := `{"traffic_splits":[{"metadata":{"name":"bar","namespace":"foo","creationTimestamp":null},"spec":{}}],"weighted_services":[{"service_name:omitempty":{"Namespace":"default","Name":"bookstore-v1"},"weight:omitempty":90,"root_service:omitempty":"bookstore-apex"},{"service_name:omitempty":{"Namespace":"default","Name":"bookstore-v2"},"weight:omitempty":10,"root_service:omitempty":"bookstore-apex"}],"service_accounts":[{"Namespace":"default","Name":"bookbuyer"}],"route_groups":[{"kind":"HTTPRouteGroup","apiVersion":"specs.smi-spec.io/v1alpha4","metadata":{"name":"bookstore-service-routes","namespace":"default","creationTimestamp":null},"spec":{"matches":[{"name":"buy-books","methods":["GET"],"pathRegex":"/buy","headers":[{"user-agent":"test-UA"}]},{"name":"sell-books","methods":["GET"],"pathRegex":"/sell","headers":[{"user-agent":"test-UA"}]},{"name":"allow-everything-on-header","headers":[{"user-agent":"test-UA"}]}]}}],"traffic_targets":[{"kind":"TrafficTarget","apiVersion":"access.smi-spec.io/v1alpha3","metadata":{"name":"bookbuyer-access-bookstore","namespace":"default","creationTimestamp":null},"spec":{"destination":{"kind":"Name","name":"bookstore","namespace":"default"},"sources":[{"kind":"Name","name":"bookbuyer","namespace":"default"}],"rules":[{"kind":"HTTPRouteGroup","name":"bookstore-service-routes","matches":["buy-books","sell-books"]}]}}]}`
+	expectedResponseBody := `{"traffic_splits":[{"metadata":{"name":"bar","namespace":"foo","creationTimestamp":null},"spec":{}}],"weighted_services":[{"service_name:omitempty":{"Namespace":"default","Name":"bookstore-v1"},"weight:omitempty":90,"root_service:omitempty":"bookstore-apex"},{"service_name:omitempty":{"Namespace":"default","Name":"bookstore-v2"},"weight:omitempty":10,"root_service:omitempty":"bookstore-apex"}],"service_accounts":[{"Namespace":"default","Name":"bookbuyer"}],"route_groups":[{"kind":"HTTPRouteGroup","apiVersion":"specs.smi-spec.io/v1alpha4","metadata":{"name":"bookstore-service-routes","namespace":"default","creationTimestamp":null},"spec":{"matches":[{"name":"buy-books","methods":["GET"],"pathRegex":"/buy","headers":[{"user-agent":"test-UA"}]},{"name":"sell-books","methods":["GET"],"pathRegex":"/sell","headers":[{"user-agent":"test-UA"}]},{"name":"allow-everything-on-header","headers":[{"user-agent":"test-UA"}]}]}}],"traffic_targets":[{"kind":"TrafficTarget","apiVersion":"access.smi-spec.io/v1alpha3","metadata":{"name":"bookbuyer-access-bookstore","namespace":"default","creationTimestamp":null},"spec":{"destination":{"kind":"ServiceAccount","name":"bookstore","namespace":"default"},"sources":[{"kind":"ServiceAccount","name":"bookbuyer","namespace":"default"}],"rules":[{"kind":"HTTPRouteGroup","name":"bookstore-service-routes","matches":["buy-books","sell-books"]}]}}]}`
 	assert.Equal(actualResponseBody, expectedResponseBody, "Actual value did not match expectations:\n%s", actualResponseBody)
 }