chore(*) add CA and Client certificate support code

Signed-off-by: Nikolay Nikolaev <nikolay.nikolaev@konghq.com>

api/mesh/v1alpha1/externalservice.proto

@@ -6,6 +6,8 @@ option go_package = "v1alpha1";
 
 import "validate/validate.proto";
 
+import "system/v1alpha1/datasource.proto";
+
 // ExternalService defines configuration of the externaly accessible service
 message ExternalService {
 
@@ -18,6 +20,15 @@ message ExternalService {
     message TLS {
       // denotes that the external service uses TLS
       bool enabled = 1;
+
+      // Data source for the certificate of CA
+      kuma.system.v1alpha1.DataSource ca_cert = 2;
+
+      // Data source for the authentication
+      kuma.system.v1alpha1.DataSource client_cert = 3;
+
+      // Data source for the authentication
+      kuma.system.v1alpha1.DataSource clinet_key = 4;
     }
 
     TLS tls = 2;
@@ -27,5 +38,5 @@ message ExternalService {
 
   // Tags associated with the external service,
   // e.g. kuma.io/service=web, kuma.io/protocol, version=1.0.
-  map<string, string> tags = 3 [ (validate.rules).map.min_pairs = 1 ];
+  map<string, string> tags = 2 [ (validate.rules).map.min_pairs = 1 ];
 }

pkg/core/runtime/builder.go

@@ -200,6 +200,7 @@ func (b *Builder) Build() (Runtime, error) {
 			ss:       b.ss,
 			cam:      b.cam,
 			xds:      b.xds,
+			dsl:      b.dsl,
 			ext:      b.ext,
 			dns:      b.dns,
 			configm:  b.configm,

pkg/core/runtime/runtime.go

@@ -3,6 +3,8 @@ package runtime
 import (
 	"context"
 
+	"github.com/kumahq/kuma/pkg/core/datasource"
+
 	"github.com/kumahq/kuma/pkg/core/dns/lookup"
 	"github.com/kumahq/kuma/pkg/core/secrets/store"
 	"github.com/kumahq/kuma/pkg/metrics"
@@ -36,6 +38,7 @@ type RuntimeInfo interface {
 type RuntimeContext interface {
 	Config() kuma_cp.Config
 	XDS() core_xds.XdsContext
+	DataSourceLoader() datasource.Loader
 	ResourceManager() core_manager.ResourceManager
 	ResourceStore() core_store.ResourceStore
 	ReadOnlyResourceManager() core_manager.ReadOnlyResourceManager
@@ -88,6 +91,7 @@ type runtimeContext struct {
 	rom      core_manager.ReadOnlyResourceManager
 	cam      ca.Managers
 	xds      core_xds.XdsContext
+	dsl      datasource.Loader
 	ext      context.Context
 	dns      dns.DNSResolver
 	configm  config_manager.ConfigManager
@@ -109,6 +113,9 @@ func (rc *runtimeContext) Config() kuma_cp.Config {
 func (rc *runtimeContext) XDS() core_xds.XdsContext {
 	return rc.xds
 }
+func (rc *runtimeContext) DataSourceLoader() datasource.Loader {
+	return rc.dsl
+}
 func (rc *runtimeContext) ResourceManager() core_manager.ResourceManager {
 	return rc.rm
 }

pkg/core/xds/types.go

@@ -44,6 +44,9 @@ type DestinationMap map[ServiceName]TagSelectorSet
 
 type ExternalService struct {
 	TLSEnabled bool
+	CaCert     []byte
+	ClientCert []byte
+	ClientKey  []byte
 }
 
 // Endpoint holds routing-related information about a single endpoint.

pkg/xds/cache/cla/cache.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/kumahq/kuma/pkg/core/datasource"
+
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/kumahq/kuma/pkg/metrics"
@@ -32,6 +34,7 @@ var (
 type Cache struct {
 	cache   *cache.Cache
 	rm      manager.ReadOnlyResourceManager
+	dsl     datasource.Loader
 	ipFunc  lookup.LookupIPFunc
 	zone    string
 	onceMap *once.Map
@@ -40,6 +43,7 @@ type Cache struct {
 
 func NewCache(
 	rm manager.ReadOnlyResourceManager,
+	dsl datasource.Loader,
 	zone string, expirationTime time.Duration,
 	ipFunc lookup.LookupIPFunc,
 	metrics metrics.Metrics,
@@ -54,6 +58,7 @@ func NewCache(
 	return &Cache{
 		cache:   cache.New(expirationTime, time.Duration(int64(float64(expirationTime)*0.9))),
 		rm:      rm,
+		dsl:     dsl,
 		zone:    zone,
 		ipFunc:  ipFunc,
 		onceMap: once.NewMap(),
@@ -85,7 +90,7 @@ func (c *Cache) GetCLA(ctx context.Context, meshName, service string) (*envoy_ap
 		if err := c.rm.List(ctx, externalServices, core_store.ListByMesh(meshName)); err != nil {
 			return nil, err
 		}
-		endpointMap := topology.BuildEndpointMap(dataplanes.Items, c.zone, mesh, externalServices.Items)
+		endpointMap := topology.BuildEndpointMap(mesh, c.zone, dataplanes.Items, externalServices.Items, c.dsl)
 		cla := endpoints.CreateClusterLoadAssignment(service, endpointMap[service])
 		c.cache.SetDefault(key, cla)
 		c.onceMap.Delete(key)

pkg/xds/cache/cla/cache_test.go

@@ -9,6 +9,11 @@ import (
 	"sync"
 	"time"
 
+	"github.com/kumahq/kuma/pkg/core/datasource"
+	"github.com/kumahq/kuma/pkg/core/secrets/cipher"
+	secret_manager "github.com/kumahq/kuma/pkg/core/secrets/manager"
+	secret_store "github.com/kumahq/kuma/pkg/core/secrets/store"
+
 	core_metrics "github.com/kumahq/kuma/pkg/metrics"
 	test_metrics "github.com/kumahq/kuma/pkg/test/metrics"
 
@@ -51,16 +56,21 @@ var _ = Describe("ClusterLoadAssignment Cache", func() {
 	expiration := 500 * time.Millisecond
 
 	BeforeEach(func() {
+		dataSourceLoader := datasource.NewDataSourceLoader(
+			secret_manager.NewSecretManager(
+				secret_store.NewSecretStore(memory.NewStore()), cipher.None(), nil))
+
 		s = memory.NewStore()
 		countingManager = &countingResourcesManager{store: s}
 		var err error
 
 		metrics, err = core_metrics.NewMetrics("Standalone")
 		Expect(err).ToNot(HaveOccurred())
 
-		claCache, err = cla.NewCache(countingManager, "", expiration, func(s string) ([]net.IP, error) {
-			return []net.IP{net.ParseIP(s)}, nil
-		}, metrics)
+		claCache, err = cla.NewCache(countingManager, dataSourceLoader, "", expiration,
+			func(s string) ([]net.IP, error) {
+				return []net.IP{net.ParseIP(s)}, nil
+			}, metrics)
 		Expect(err).ToNot(HaveOccurred())
 	})
 

pkg/xds/server/components.go

@@ -203,7 +203,8 @@ func DefaultDataplaneSyncTracker(rt core_runtime.Runtime, reconciler, ingressRec
 	if err != nil {
 		return nil, err
 	}
-	claCache, err := cla.NewCache(rt.ReadOnlyResourceManager(), rt.Config().Multicluster.Remote.Zone,
+	claCache, err := cla.NewCache(
+		rt.ReadOnlyResourceManager(), rt.DataSourceLoader(), rt.Config().Multicluster.Remote.Zone,
 		rt.Config().Store.Cache.ExpirationTime, rt.LookupIP(), rt.Metrics())
 	if err != nil {
 		return nil, err
@@ -301,7 +302,9 @@ func DefaultDataplaneSyncTracker(rt core_runtime.Runtime, reconciler, ingressRec
 				destinations := xds_topology.BuildDestinationMap(dataplane, routes)
 
 				// resolve all endpoints that match given selectors
-				outbound := xds_topology.BuildEndpointMap(dataplanes.Items, rt.Config().Multicluster.Remote.Zone, mesh, externalServices.Items)
+				outbound := xds_topology.BuildEndpointMap(
+					mesh, rt.Config().Multicluster.Remote.Zone,
+					dataplanes.Items, externalServices.Items, rt.DataSourceLoader())
 
 				healthChecks, err := xds_topology.GetHealthChecks(ctx, dataplane, destinations, rt.ReadOnlyResourceManager())
 				if err != nil {