draft: support passing Secret source over MCP

pilot/pkg/bootstrap/server.go

@@ -504,14 +504,17 @@ func (s *Server) WaitUntilCompletion() {
 
 // initSDSServer starts the SDS server
 func (s *Server) initSDSServer() {
-	if s.kubeClient == nil {
+	if s.kubeClient == nil && !features.EnableSecretOverMCP {
 		return
 	}
-	if !features.EnableXDSIdentityCheck {
+	if !features.EnableXDSIdentityCheck && !features.EnableSecretOverMCP {
 		// Make sure we have security
 		log.Warnf("skipping Kubernetes credential reader; PILOT_ENABLE_XDS_IDENTITY_CHECK must be set to true for this feature.")
-	} else {
-		creds := kubecredentials.NewMulticluster(s.clusterID)
+		return
+	}
+	var creds *kubecredentials.Multicluster
+	if s.kubeClient != nil {
+		creds = kubecredentials.NewMulticluster(s.clusterID)
 		creds.AddSecretHandler(func(name string, namespace string) {
 			s.XDSServer.ConfigUpdate(&model.PushRequest{
 				Full: false,
@@ -525,12 +528,12 @@ func (s *Server) initSDSServer() {
 				Reason: []model.TriggerReason{model.SecretTrigger},
 			})
 		})
-		s.XDSServer.Generators[v3.SecretType] = xds.NewSecretGen(creds, s.XDSServer.Cache, s.clusterID, s.environment.Mesh())
 		s.multiclusterController.AddHandler(creds)
 		if ecdsGen, found := s.XDSServer.Generators[v3.ExtensionConfigurationType]; found {
 			ecdsGen.(*xds.EcdsGenerator).SetCredController(creds)
 		}
 	}
+	s.XDSServer.Generators[v3.SecretType] = xds.NewSecretGen(creds, s.XDSServer.Cache, s.clusterID, s.environment.Mesh())
 }
 
 // initKubeClient creates the k8s client if running in an k8s environment.

pilot/pkg/credentials/kube/secrets.go

@@ -182,7 +182,7 @@ func (s *CredentialsController) GetKeyAndCert(name, namespace string) (key []byt
 		return nil, nil, fmt.Errorf("secret %v/%v not found", namespace, name)
 	}
 
-	return extractKeyAndCert(k8sSecret)
+	return ExtractKeyAndCert(k8sSecret)
 }
 
 func (s *CredentialsController) GetCaCert(name, namespace string) (cert []byte, err error) {
@@ -194,9 +194,9 @@ func (s *CredentialsController) GetCaCert(name, namespace string) (cert []byte,
 		if caCertErr != nil {
 			return nil, fmt.Errorf("secret %v/%v not found", namespace, strippedName)
 		}
-		return extractRoot(k8sSecret)
+		return ExtractRoot(k8sSecret)
 	}
-	return extractRoot(k8sSecret)
+	return ExtractRoot(k8sSecret)
 }
 
 func (s *CredentialsController) GetDockerCredential(name, namespace string) ([]byte, error) {
@@ -233,8 +233,8 @@ func hasValue(d map[string][]byte, keys ...string) bool {
 	return true
 }
 
-// extractKeyAndCert extracts server key, certificate
-func extractKeyAndCert(scrt *v1.Secret) (key, cert []byte, err error) {
+// ExtractKeyAndCert extracts server key, certificate
+func ExtractKeyAndCert(scrt *v1.Secret) (key, cert []byte, err error) {
 	if hasValue(scrt.Data, GenericScrtCert, GenericScrtKey) {
 		return scrt.Data[GenericScrtKey], scrt.Data[GenericScrtCert], nil
 	}
@@ -265,8 +265,8 @@ func truncatedKeysMessage(data map[string][]byte) string {
 	return fmt.Sprintf("%s, and %d more...", strings.Join(keys[:3], ", "), len(keys)-3)
 }
 
-// extractRoot extracts the root certificate
-func extractRoot(scrt *v1.Secret) (cert []byte, err error) {
+// ExtractRoot extracts the root certificate
+func ExtractRoot(scrt *v1.Secret) (cert []byte, err error) {
 	if hasValue(scrt.Data, GenericScrtCaCert) {
 		return scrt.Data[GenericScrtCaCert], nil
 	}

pilot/pkg/features/pilot.go

@@ -406,6 +406,12 @@ var (
 		"If enabled, pilot will authorize XDS clients, to ensure they are acting only as namespaces they have permissions for.",
 	).Get()
 
+	EnableSecretOverMCP = env.RegisterBoolVar(
+		"PILOT_ENABLE_SECRET_OVER_MCP",
+		true,
+		"If enabled, pilot will accept secret resouces over MCP",
+	).Get()
+
 	// TODO: Move this to proper API.
 	trustedGatewayCIDR = env.RegisterStringVar(
 		"TRUSTED_GATEWAY_CIDR",

pilot/pkg/model/push_context.go

@@ -16,6 +16,7 @@ package model
 
 import (
 	"encoding/json"
+	v1 "k8s.io/api/core/v1"
 	"math"
 	"sort"
 	"strings"
@@ -197,6 +198,9 @@ type PushContext struct {
 	// gatewayIndex is the index of gateways.
 	gatewayIndex gatewayIndex
 
+	// secrets for each namespace
+	SecretsByNameSpace map[string]map[string]config.Config
+
 	// clusterLocalHosts extracted from the MeshConfig
 	clusterLocalHosts ClusterLocalHosts
 
@@ -647,6 +651,7 @@ func NewPushContext() *PushContext {
 		sidecarIndex:            newSidecarIndex(),
 		envoyFiltersByNamespace: map[string][]*EnvoyFilterWrapper{},
 		gatewayIndex:            newGatewayIndex(),
+		SecretsByNameSpace:      map[string]map[string]config.Config{},
 		ProxyStatus:             map[string]map[string]ProxyPushStatus{},
 		ServiceAccounts:         map[host.Name]map[int][]string{},
 	}
@@ -1164,6 +1169,10 @@ func (ps *PushContext) createNewContext(env *Environment) error {
 		return err
 	}
 
+	if err := ps.initSecrets(env); err != nil {
+		return err
+	}
+
 	if err := ps.initKubernetesGateways(env); err != nil {
 		return err
 	}
@@ -1219,12 +1228,14 @@ func (ps *PushContext) updateContext(
 ) error {
 	var servicesChanged, virtualServicesChanged, destinationRulesChanged, gatewayChanged,
 		authnChanged, authzChanged, envoyFiltersChanged, sidecarsChanged, telemetryChanged, gatewayAPIChanged,
-		wasmPluginsChanged, proxyConfigsChanged bool
+		wasmPluginsChanged, proxyConfigsChanged, secretChanged bool
 
 	for conf := range pushReq.ConfigsUpdated {
 		switch conf.Kind {
 		case kind.ServiceEntry:
 			servicesChanged = true
+		case kind.Secret:
+			secretChanged = true
 		case kind.DestinationRule:
 			destinationRulesChanged = true
 		case kind.VirtualService:
@@ -1265,6 +1276,12 @@ func (ps *PushContext) updateContext(
 		ps.ServiceAccounts = oldPushContext.ServiceAccounts
 	}
 
+	if secretChanged {
+		if err := ps.initSecrets(env); err != nil {
+			return err
+		}
+	}
+
 	if servicesChanged || gatewayAPIChanged {
 		// Gateway status depends on services, so recompute if they change as well
 		if err := ps.initKubernetesGateways(env); err != nil {
@@ -1416,6 +1433,22 @@ func (ps *PushContext) initServiceRegistry(env *Environment) error {
 	return nil
 }
 
+// pre computes secrets per namespace
+func (ps *PushContext) initSecrets(env *Environment) error {
+	secretConfigs, err := env.List(gvk.Secret, NamespaceAll)
+	if err != nil {
+		return err
+	}
+	sortConfigByCreationTime(secretConfigs)
+	for _, secretConfig := range secretConfigs {
+		if _, exists := ps.SecretsByNameSpace[secretConfig.Namespace]; !exists {
+			ps.SecretsByNameSpace[secretConfig.Namespace] = map[string]config.Config{}
+		}
+		ps.SecretsByNameSpace[secretConfig.Namespace][secretConfig.Name] = secretConfig
+	}
+	return nil
+}
+
 // SortServicesByCreationTime sorts the list of services in ascending order by their creation time (if available).
 func SortServicesByCreationTime(services []*Service) []*Service {
 	sort.SliceStable(services, func(i, j int) bool {
@@ -2063,6 +2096,15 @@ func (ps *PushContext) NetworkManager() *NetworkManager {
 	return ps.networkMgr
 }
 
+func (ps *PushContext) GetSecret(name, namespace string) *v1.Secret {
+	if m, exists := ps.SecretsByNameSpace[namespace]; exists {
+		if config, exists := m[name]; exists {
+			return config.Spec.(*v1.Secret)
+		}
+	}
+	return nil
+}
+
 // BestEffortInferServiceMTLSMode infers the mTLS mode for the service + port from all authentication
 // policies (both alpha and beta) in the system. The function always returns MTLSUnknown for external service.
 // The result is a best effort. It is because the PeerAuthentication is workload-based, this function is unable

pilot/pkg/xds/sds.go

@@ -18,6 +18,7 @@ import (
 	"crypto/x509"
 	"encoding/pem"
 	"fmt"
+	"istio.io/istio/pilot/pkg/credentials/kube"
 	"strings"
 	"time"
 
@@ -91,7 +92,107 @@ func (s *SecretGen) parseResources(names []string, proxy *model.Proxy) []SecretR
 	return res
 }
 
+func getCaCert(ps *model.PushContext, name, namespace string) (cert []byte, err error) {
+	if secret := ps.GetSecret(name, namespace); secret != nil {
+		return kube.ExtractRoot(secret)
+	}
+	// Could not fetch cert, look for secret without -cacert suffix
+	strippedName := strings.TrimSuffix(name, securitymodel.SdsCaSuffix)
+	if secret := ps.GetSecret(strippedName, namespace); secret != nil {
+		return kube.ExtractRoot(secret)
+	}
+	return nil, fmt.Errorf("secret %v/%v not found", namespace, strippedName)
+}
+
+func getKeyAndCert(ps *model.PushContext, name, namespace string) (key []byte, cert []byte, err error) {
+	if secret := ps.GetSecret(name, namespace); secret != nil {
+		return kube.ExtractKeyAndCert(secret)
+	}
+	return nil, nil, fmt.Errorf("secret %v/%v not found", namespace, name)
+}
+
+func (s *SecretGen) generateFromPushContext(sr SecretResource, proxy *model.Proxy, ps *model.PushContext) *discovery.Resource {
+	isCAOnlySecret := strings.HasSuffix(sr.Name, securitymodel.SdsCaSuffix)
+	if isCAOnlySecret {
+		caCert, err := getCaCert(ps, sr.Name, sr.Namespace)
+		if err != nil {
+			pilotSDSCertificateErrors.Increment()
+			log.Warnf("failed to fetch ca certificate for %s: %v", sr.ResourceName, err)
+			return nil
+		}
+		if features.VerifySDSCertificate {
+			if err := validateCertificate(caCert); err != nil {
+				recordInvalidCertificate(sr.ResourceName, err)
+				return nil
+			}
+		}
+		res := toEnvoyCaSecret(sr.ResourceName, caCert)
+		return res
+	}
+
+	key, cert, err := getKeyAndCert(ps, sr.Name, sr.Namespace)
+	if err != nil {
+		pilotSDSCertificateErrors.Increment()
+		log.Warnf("failed to fetch key and certificate for %s: %v", sr.ResourceName, err)
+		return nil
+	}
+	if features.VerifySDSCertificate {
+		if err := validateCertificate(cert); err != nil {
+			recordInvalidCertificate(sr.ResourceName, err)
+			return nil
+		}
+	}
+	res := toEnvoyKeyCertSecret(sr.ResourceName, key, cert, proxy, s.meshConfig)
+	return res
+
+}
+
+func (s *SecretGen) generateMCP(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
+	if req == nil || !sdsNeedsPush(req.ConfigsUpdated) {
+		return nil, model.DefaultXdsLogDetails, nil
+	}
+	var updatedSecrets map[model.ConfigKey]struct{}
+	if !req.Full {
+		updatedSecrets = model.ConfigsOfKind(req.ConfigsUpdated, kind.Secret)
+	}
+
+	resources := s.parseResources(w.ResourceNames, proxy)
+
+	results := model.Resources{}
+	cached, regenerated := 0, 0
+	for _, sr := range resources {
+		if updatedSecrets != nil {
+			if !containsAny(updatedSecrets, relatedConfigs(model.ConfigKey{Kind: kind.Secret, Name: sr.Name, Namespace: sr.Namespace})) {
+				// This is an incremental update, filter out secrets that are not updated.
+				continue
+			}
+		}
+
+		cachedItem, f := s.cache.Get(sr)
+		if f && !features.EnableUnsafeAssertions {
+			// If it is in the Cache, add it and continue
+			// We skip cache if assertions are enabled, so that the cache will assert our eviction logic is correct
+			results = append(results, cachedItem)
+			cached++
+			continue
+		}
+		regenerated++
+		res := s.generateFromPushContext(sr, proxy, req.Push)
+		if res != nil {
+			s.cache.Add(sr, req, res)
+			results = append(results, res)
+		}
+	}
+	return results, model.XdsLogDetails{
+		Incremental:    updatedSecrets != nil,
+		AdditionalInfo: fmt.Sprintf("cached:%v/%v", cached, cached+regenerated),
+	}, nil
+}
+
 func (s *SecretGen) Generate(proxy *model.Proxy, w *model.WatchedResource, req *model.PushRequest) (model.Resources, model.XdsLogDetails, error) {
+	if s.secrets == nil {
+		return s.generateMCP(proxy, w, req)
+	}
 	if proxy.VerifiedIdentity == nil {
 		log.Warnf("proxy %s is not authorized to receive credscontroller. Ensure you are connecting over TLS port and are authenticated.", proxy.ID)
 		return nil, model.DefaultXdsLogDetails, nil

pkg/config/schema/metadata.yaml

@@ -140,6 +140,7 @@ collections:
     kind: "Secret"
     group: ""
     builtin: true
+    pilot: true
 
   - name: "k8s/core/v1/services"
     kind: "Service"