Remove certificate common name from the proxy struct.

The CN is no longer static for the duration of the proxy,
and is not required to key the proxy.

Signed-off-by: Sean Teeling <[email protected]>

Author: steeling

pkg/catalog/fake/fake.go

@@ -27,15 +27,9 @@ import (
 
 // NewFakeMeshCatalog creates a new struct implementing catalog.MeshCataloger interface used for testing.
 func NewFakeMeshCatalog(kubeClient kubernetes.Interface, meshConfigClient configClientset.Interface) *catalog.MeshCatalog {
-	var (
-		mockCtrl             *gomock.Controller
-		mockKubeController   *k8s.MockController
-		mockPolicyController *policy.MockController
-	)
-
-	mockCtrl = gomock.NewController(ginkgo.GinkgoT())
-	mockKubeController = k8s.NewMockController(mockCtrl)
-	mockPolicyController = policy.NewMockController(mockCtrl)
+	mockCtrl := gomock.NewController(ginkgo.GinkgoT())
+	mockKubeController := k8s.NewMockController(mockCtrl)
+	mockPolicyController := policy.NewMockController(mockCtrl)
 
 	meshSpec := smiFake.NewFakeMeshSpecClient()
 

pkg/catalog/traffictarget.go

@@ -139,8 +139,7 @@ func trafficTargetIdentityToSvcAccount(identitySubject smiAccess.IdentityBinding
 
 // trafficTargetIdentityToServiceIdentity returns an identity of the form <namespace>/<service-account>
 func trafficTargetIdentityToServiceIdentity(identitySubject smiAccess.IdentityBindingSubject) identity.ServiceIdentity {
-	svcAccount := trafficTargetIdentityToSvcAccount(identitySubject)
-	return identity.GetKubernetesServiceIdentity(svcAccount, identity.ClusterLocalTrustDomain)
+	return trafficTargetIdentityToSvcAccount(identitySubject).ToServiceIdentity()
 }
 
 // trafficTargetIdentitiesToSvcAccounts returns a list of Service Accounts from the given list of identities from a Traffic Target

pkg/debugger/proxy.go

@@ -51,8 +51,8 @@ func (ds DebugConfig) printProxies(w http.ResponseWriter) {
 		ts := proxy.GetConnectedAt()
 		proxyURL := fmt.Sprintf("/debug/proxy?%s=%s", uuidQueryKey, proxy.UUID)
 		configDumpURL := fmt.Sprintf("%s&%s=%t", proxyURL, proxyConfigQueryKey, true)
-		_, _ = fmt.Fprintf(w, `<tr><td>%d:</td><td>%s</td><td>%+v</td><td>(%+v ago)</td><td><a href="%s">certs</a></td><td><a href="%s">cfg</a></td></tr>`,
-			idx+1, proxy.Identity, ts, time.Since(ts), proxyURL, configDumpURL)
+		_, _ = fmt.Fprintf(w, `<tr><td>%d:</td><td>%s</td><td>%s</td><td>%+v</td><td>(%+v ago)</td><td><a href="%s">certs</a></td><td><a href="%s">cfg</a></td></tr>`,
+			idx+1, proxy.Identity, proxy.UUID, ts, time.Since(ts), proxyURL, configDumpURL)
 	}
 	_, _ = fmt.Fprint(w, `</table>`)
 }
@@ -67,9 +67,9 @@ func (ds DebugConfig) getConfigDump(uuid string, w http.ResponseWriter) {
 		}
 		return
 	}
-	pod, err := envoy.GetPodFromCertificate(proxy.GetCertificateCommonName(), ds.kubeController)
+	pod, err := ds.kubeController.GetPodForProxy(proxy)
 	if err != nil {
-		msg := fmt.Sprintf("Error getting Pod from certificate with CN=%s", proxy.GetCertificateCommonName())
+		msg := fmt.Sprintf("Error getting Pod from proxy %s", proxy.GetName())
 		log.Error().Err(err).Msg(msg)
 		if _, err := w.Write([]byte(msg)); err != nil {
 			log.Error().Err(err).Msg("Error writing debugger response")
@@ -91,9 +91,9 @@ func (ds DebugConfig) getProxy(uuid string, w http.ResponseWriter) {
 		}
 		return
 	}
-	pod, err := envoy.GetPodFromCertificate(proxy.GetCertificateCommonName(), ds.kubeController)
+	pod, err := ds.kubeController.GetPodForProxy(proxy)
 	if err != nil {
-		msg := fmt.Sprintf("Error getting Pod from certificate with CN=%s", proxy.GetCertificateCommonName())
+		msg := fmt.Sprintf("Error getting Pod from proxy %s", proxy.GetName())
 		log.Error().Err(err).Msg(msg)
 		if _, err := w.Write([]byte(msg)); err != nil {
 			log.Error().Err(err).Msg("Error writing debugger response")

pkg/envoy/ads/cache_stream.go

@@ -12,10 +12,10 @@ import (
 	v1 "k8s.io/api/core/v1"
 
 	"github.com/openservicemesh/osm/pkg/announcements"
-	"github.com/openservicemesh/osm/pkg/certificate"
 	"github.com/openservicemesh/osm/pkg/constants"
 	"github.com/openservicemesh/osm/pkg/envoy"
 	"github.com/openservicemesh/osm/pkg/errcode"
+	"github.com/openservicemesh/osm/pkg/identity"
 )
 
 // Routine which fulfills listening to proxy broadcasts
@@ -61,9 +61,6 @@ func (s *Server) allPodUpdater() {
 // All verticals use the proxy structure to infer the pod later, so the actual only mandatory
 // data for the verticals to be functional is the common name, which links proxy <-> pod
 func GetProxyFromPod(pod *v1.Pod) (*envoy.Proxy, error) {
-	var serviceAccount string
-	var namespace string
-
 	uuidString, uuidFound := pod.Labels[constants.EnvoyUniqueIDLabelName]
 	if !uuidFound {
 		return nil, errors.Errorf("UUID not found for pod %s/%s, not a mesh pod", pod.Namespace, pod.Name)
@@ -73,27 +70,18 @@ func GetProxyFromPod(pod *v1.Pod) (*envoy.Proxy, error) {
 		return nil, errors.Errorf("Could not parse UUID label into UUID type (%s): %v", uuidString, err)
 	}
 
-	serviceAccount = pod.Spec.ServiceAccountName
-	namespace = pod.Namespace
-
-	// construct CN for this pod/proxy
-	// TODO: Infer proxy type from Pod
-	commonName := envoy.NewXDSCertCommonName(proxyUUID, envoy.KindSidecar, serviceAccount, namespace)
-	tempProxy, err := envoy.NewProxy(certificate.CommonName(commonName), "NoSerial", &net.IPAddr{IP: net.IPv4zero})
+	sa := pod.Spec.ServiceAccountName
+	namespace := pod.Namespace
 
-	return tempProxy, err
+	return envoy.NewProxy(envoy.KindSidecar, proxyUUID, identity.New(sa, namespace), &net.IPAddr{IP: net.IPv4zero}), nil
 }
 
 // RecordFullSnapshot stores a group of resources as a new Snapshot with a new version in the cache.
 // It also runs a consistency check on the snapshot (will warn if there are missing resources referenced in
 // the snapshot)
 func (s *Server) RecordFullSnapshot(proxy *envoy.Proxy, snapshotResources map[string][]types.Resource) error {
-	s.configVerMutex.Lock()
-	s.configVersion[proxy.GetCertificateCommonName().String()]++
-	s.configVerMutex.Unlock()
-
 	snapshot, err := cache.NewSnapshot(
-		fmt.Sprintf("%d", s.configVersion[proxy.GetCertificateCommonName().String()]),
+		fmt.Sprintf("%d", s.configVersion[proxy.UUID.String()]),
 		snapshotResources,
 	)
 	if err != nil {
@@ -104,5 +92,9 @@ func (s *Server) RecordFullSnapshot(proxy *envoy.Proxy, snapshotResources map[st
 		log.Warn().Err(err).Str("proxy", proxy.String()).Msgf("Snapshot for proxy not consistent")
 	}
 
-	return s.ch.SetSnapshot(context.TODO(), proxy.GetCertificateCommonName().String(), snapshot)
+	s.configVerMutex.Lock()
+	defer s.configVerMutex.Unlock()
+	s.configVersion[proxy.UUID.String()]++
+
+	return s.ch.SetSnapshot(context.TODO(), proxy.UUID.String(), snapshot)
 }

pkg/envoy/ads/cache_test.go

@@ -8,21 +8,18 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
-	"github.com/openservicemesh/osm/pkg/certificate"
 	"github.com/openservicemesh/osm/pkg/constants"
-	"github.com/openservicemesh/osm/pkg/envoy"
 )
 
 func TestGetProxyFromPod(t *testing.T) {
 	assert := tassert.New(t)
 
 	var (
 		// Default fixtures for various test variables
-		podName         = "pod"
-		namespace       = "namespace"
-		serviceAccount  = "serviceAccount"
-		validUUID       = uuid.New()
-		validCommonName = envoy.NewXDSCertCommonName(validUUID, envoy.KindSidecar, serviceAccount, namespace)
+		podName        = "pod"
+		namespace      = "namespace"
+		serviceAccount = "serviceAccount"
+		validUUID      = uuid.New()
 	)
 
 	testCases := []struct {
@@ -32,8 +29,7 @@ func TestGetProxyFromPod(t *testing.T) {
 		pod *v1.Pod
 
 		// Output check
-		expectErr  bool
-		commonName certificate.CommonName
+		expectErr bool
 	}{
 		{
 			testCaseName: "Pod with no label",
@@ -80,7 +76,6 @@ func TestGetProxyFromPod(t *testing.T) {
 					ServiceAccountName: serviceAccount,
 				},
 			},
-			commonName: validCommonName,
 		},
 	}
 
@@ -90,8 +85,8 @@ func TestGetProxyFromPod(t *testing.T) {
 		if testCase.expectErr {
 			assert.Error(err)
 		} else {
-			assert.Equal(proxyResult.GetCertificateCommonName(), testCase.commonName,
-				"%s: Did not return equal common name")
+			assert.NotNil(proxyResult)
+			assert.Equal(proxyResult.UUID, validUUID)
 		}
 	}
 }

pkg/envoy/ads/errors.go

@@ -10,3 +10,4 @@ var errGrpcClosed = errors.New("grpc closed")
 var errTooManyConnections = errors.New("too many connections")
 var errServiceAccountMismatch = errors.New("service account mismatch in nodeid vs xds certificate common name")
 var errUnsuportedXDSRequest = errors.New("Unsupported XDS server connection type")
+var errInvalidCertificateCN = errors.New("invalid cn")

pkg/envoy/ads/jobs.go

@@ -37,12 +37,5 @@ func (proxyJob *proxyResponseJob) Run() {
 
 // JobName implementation for this job, for logging purposes
 func (proxyJob *proxyResponseJob) JobName() string {
-	return fmt.Sprintf("sendJob-%s", proxyJob.proxy.GetCertificateSerialNumber())
-}
-
-// Hash implementation for this job to hash into the worker queues
-func (proxyJob *proxyResponseJob) Hash() uint64 {
-	// Uses proxy hash to always serialize work for the same proxy to the same worker,
-	// this avoid out-of-order mishandling of envoy updates by multiple workers
-	return proxyJob.proxy.GetHash()
+	return fmt.Sprintf("sendJob-%s", proxyJob.proxy.GetName())
 }

pkg/envoy/ads/profile_test.go

@@ -1,7 +1,6 @@
 package ads
 
 import (
-	"fmt"
 	"testing"
 
 	xds_cluster "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
@@ -10,15 +9,14 @@ import (
 	"github.com/google/uuid"
 	tassert "github.com/stretchr/testify/assert"
 
-	"github.com/openservicemesh/osm/pkg/certificate"
 	"github.com/openservicemesh/osm/pkg/envoy"
+	"github.com/openservicemesh/osm/pkg/identity"
 	"github.com/openservicemesh/osm/pkg/tests"
 )
 
 func TestValidateResourcesRequestResponse(t *testing.T) {
 	assert := tassert.New(t)
-	proxy, err := envoy.NewProxy(certificate.CommonName(fmt.Sprintf("%s.sidecar.foo.bar", uuid.New())), certificate.SerialNumber("123"), tests.NewMockAddress("1.2.3.4"))
-	assert.Nil(err)
+	proxy := envoy.NewProxy(envoy.KindSidecar, uuid.New(), identity.New("foo", "bar"), tests.NewMockAddress("1.2.3.4"))
 
 	testCases := []struct {
 		request          *xds_discovery.DiscoveryRequest

pkg/envoy/ads/response.go

@@ -132,7 +132,7 @@ func (s *Server) SendDiscoveryResponse(proxy *envoy.Proxy, request *xds_discover
 		proto, err := anypb.New(res.(proto.Message))
 		if err != nil {
 			log.Error().Err(err).Str(errcode.Kind, errcode.GetErrCodeWithMetric(errcode.ErrMarshallingXDSResource)).
-				Msgf("Error marshalling resource %s for proxy %s", typeURI, proxy.GetCertificateSerialNumber())
+				Msgf("Error marshalling resource %s for proxy %s", typeURI, proxy.GetName())
 			continue
 		}
 		// Add resource to response
@@ -161,15 +161,15 @@ func (s *Server) SendDiscoveryResponse(proxy *envoy.Proxy, request *xds_discover
 
 	// Send the response
 	if err := (*server).Send(response); err != nil {
-		metricsstore.DefaultMetricsStore.ProxyResponseSendErrorCount.WithLabelValues(proxy.GetCertificateCommonName().String(), string(typeURI)).Inc()
+		metricsstore.DefaultMetricsStore.ProxyResponseSendErrorCount.WithLabelValues(proxy.GetName(), string(typeURI)).Inc()
 		log.Error().Err(err).Str(errcode.Kind, errcode.GetErrCodeWithMetric(errcode.ErrSendingDiscoveryResponse)).
 			Str("proxy", proxy.String()).Msgf("Error sending response for typeURI %s to proxy", typeURI.Short())
 		return err
 	}
 
 	// Sending discovery response succeeded, record last resources sent
 	proxy.SetLastResourcesSent(typeURI, resourcesSent)
-	metricsstore.DefaultMetricsStore.ProxyResponseSendSuccessCount.WithLabelValues(proxy.GetCertificateCommonName().String(), string(typeURI)).Inc()
+	metricsstore.DefaultMetricsStore.ProxyResponseSendSuccessCount.WithLabelValues(proxy.GetName(), string(typeURI)).Inc()
 
 	return nil
 }

pkg/envoy/ads/response_test.go

@@ -87,21 +87,17 @@ var _ = Describe("Test ADS response functions", func() {
 		GinkgoT().Fatalf("Error creating new Bookstire Apex service: %s", err.Error())
 	}
 
-	certCommonName := envoy.NewXDSCertCommonName(proxyUUID, envoy.KindSidecar, proxySvcAccount.Name, proxySvcAccount.Namespace)
-	certSerialNumber := certificate.SerialNumber("123456")
-	proxy, err := envoy.NewProxy(certCommonName, certSerialNumber, nil)
+	proxy := envoy.NewProxy(envoy.KindSidecar, proxyUUID, proxySvcAccount.ToServiceIdentity(), nil)
 
 	Context("Proxy is valid", func() {
 		Expect(proxy).ToNot((BeNil()))
-		Expect(err).ToNot(HaveOccurred())
 	})
 
 	Context("Test sendAllResponses()", func() {
 
 		certManager := tresorFake.NewFake(nil)
-		certCommonName := certificate.CommonName(fmt.Sprintf("%s.%s.cluster.local", proxySvcAccount.Name, proxySvcAccount.Namespace))
 		certDuration := 1 * time.Hour
-		certPEM, _ := certManager.IssueCertificate(certCommonName, certDuration)
+		certPEM, _ := certManager.IssueCertificate(certificate.CommonName(proxySvcAccount.ToServiceIdentity().String()), certDuration)
 		cert, _ := certificate.DecodePEMCertificate(certPEM.GetCertificateChain())
 		server, actualResponses := tests.NewFakeXDSServer(cert, nil, nil)
 		kubectrlMock := k8s.NewMockController(mockCtrl)
@@ -118,6 +114,8 @@ var _ = Describe("Test ADS response functions", func() {
 		}).AnyTimes()
 		mockConfigurator.EXPECT().GetMeshConfig().AnyTimes()
 
+		mc.GetKubeController().(*k8s.MockController).EXPECT().GetPodForProxy(proxy).Return(&pod, nil).AnyTimes()
+
 		metricsstore.DefaultMetricsStore.Start(metricsstore.DefaultMetricsStore.ProxyResponseSendSuccessCount)
 
 		It("returns Aggregated Discovery Service response", func() {
@@ -174,7 +172,7 @@ var _ = Describe("Test ADS response functions", func() {
 				CertType: secrets.ServiceCertType,
 			}.String()))
 
-			Expect(metricsstore.DefaultMetricsStore.Contains(fmt.Sprintf("osm_proxy_response_send_success_count{common_name=%q,type=%q} 1\n", proxy.GetCertificateCommonName(), envoy.TypeCDS))).To(BeTrue())
+			Expect(metricsstore.DefaultMetricsStore.Contains(fmt.Sprintf("osm_proxy_response_send_success_count{proxy_name=%q,type=%q} 1\n", proxy.GetName(), envoy.TypeCDS))).To(BeTrue())
 		})
 	})
 

pkg/envoy/ads/stream.go

@@ -8,6 +8,7 @@ import (
 
 	mapset "github.com/deckarep/golang-set"
 	xds_discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
+	"github.com/google/uuid"
 	"github.com/pkg/errors"
 
 	"github.com/openservicemesh/osm/pkg/announcements"
@@ -33,24 +34,23 @@ func (s *Server) StreamAggregatedResources(server xds_discovery.AggregatedDiscov
 	}
 
 	// If maxDataPlaneConnections is enabled i.e. not 0, then check that the number of Envoy connections is less than maxDataPlaneConnections
-	if s.cfg.GetMaxDataPlaneConnections() > 0 && s.proxyRegistry.GetConnectedProxyCount() >= s.cfg.GetMaxDataPlaneConnections() {
+	if s.cfg.GetMaxDataPlaneConnections() != 0 && s.proxyRegistry.GetConnectedProxyCount() >= s.cfg.GetMaxDataPlaneConnections() {
 		metricsstore.DefaultMetricsStore.ProxyMaxConnectionsRejected.Inc()
 		return errTooManyConnections
 	}
 
 	log.Trace().Msgf("Envoy with certificate SerialNumber=%s connected", certSerialNumber)
 	metricsstore.DefaultMetricsStore.ProxyConnectCount.Inc()
 
+	kind, uuid, si, err := getCertificateCommonNameMeta(certCommonName)
+	if err != nil {
+		return fmt.Errorf("error parsing certificate common name %s: %w", certCommonName, err)
+	}
+
 	// This is the Envoy proxy that just connected to the control plane.
 	// NOTE: This is step 1 of the registration. At this point we do not yet have context on the Pod.
 	//       Details on which Pod this Envoy is fronting will arrive via xDS in the NODE_ID string.
-	//       When this arrives we will call RegisterProxy() a second time - this time with Pod context!
-	proxy, err := envoy.NewProxy(certCommonName, certSerialNumber, utils.GetIPFromContext(server.Context()))
-	if err != nil {
-		log.Error().Err(err).Str(errcode.Kind, errcode.GetErrCodeWithMetric(errcode.ErrInitializingProxy)).
-			Msgf("Error initializing proxy with certificate SerialNumber=%s", certSerialNumber)
-		return err
-	}
+	proxy := envoy.NewProxy(kind, uuid, si, utils.GetIPFromContext(server.Context()))
 
 	if err := s.recordPodMetadata(proxy); err == errServiceAccountMismatch {
 		// Service Account mismatch
@@ -332,6 +332,22 @@ func isCNforProxy(proxy *envoy.Proxy, cn certificate.CommonName) bool {
 	return identityForCN == proxy.Identity.ToK8sServiceAccount()
 }
 
+func getCertificateCommonNameMeta(cn certificate.CommonName) (envoy.ProxyKind, uuid.UUID, identity.ServiceIdentity, error) {
+	// XDS cert CN is of the form <proxy-UUID>.<kind>.<proxy-identity>.<namespace>.<trust-domain>
+	chunks := strings.SplitN(cn.String(), constants.DomainDelimiter, 5)
+	if len(chunks) < 3 {
+		return "", uuid.UUID{}, "", errInvalidCertificateCN
+	}
+	proxyUUID, err := uuid.Parse(chunks[0])
+	if err != nil {
+		log.Error().Err(err).Str(errcode.Kind, errcode.GetErrCodeWithMetric(errcode.ErrParsingXDSCertCN)).
+			Msgf("Error parsing %s into uuid.UUID", chunks[0])
+		return "", uuid.UUID{}, "", err
+	}
+
+	return envoy.ProxyKind(chunks[1]), proxyUUID, identity.New(chunks[2], chunks[3]), nil
+}
+
 // recordPodMetadata records pod metadata and verifies the certificate issued for this pod
 // is for the same service account as seen on the pod's service account
 func (s *Server) recordPodMetadata(p *envoy.Proxy) error {
@@ -341,7 +357,7 @@ func (s *Server) recordPodMetadata(p *envoy.Proxy) error {
 		return nil
 	}
 
-	pod, err := envoy.GetPodFromCertificate(p.GetCertificateCommonName(), s.kubecontroller)
+	pod, err := s.kubecontroller.GetPodForProxy(p)
 	if err != nil {
 		log.Warn().Str("proxy", p.String()).Msg("Could not find pod for connecting proxy. No metadata was recorded.")
 		return nil