第四章 SDS Server

type Server struct {
	workloadSds *sdsservice
	gatewaySds  *sdsservice
    ...
}

type SecretManager interface {
	GenerateSecret(ctx context.Context, connectionID, resourceName, token string) (*model.SecretItem, error)
	SecretExist(connectionID, resourceName, token, version string) bool
	DeleteSecret(connectionID, resourceName string)
    ...
}

type sdsservice struct {
	st cache.SecretManager
    ...
}

type SecretCache struct {
	fetcher        *secretfetcher.SecretFetcher
    ...
}

type Client interface {
	CSRSign(ctx context.Context, reqID string, csrPEM []byte, subjectID string,
		certValidTTLInSec int64) ([]string /*PEM-encoded certificate chain*/, error)
}

type SecretFetcher struct {
    ...
	CaClient    caClientInterface.Client
    ...
}

type SecretCache struct {
    ...
	fetcher        *secretfetcher.SecretFetcher
    ...
	// callback function to invoke when detecting secret change.
	notifyCallback func(connKey ConnKey, secret *model.SecretItem) error
    ...
}

	proxyCmd = &cobra.Command{
		Use:   "proxy",
		Short: "Envoy proxy agent",
		FParseErrWhitelist: cobra.FParseErrWhitelist{
			// Allow unknown flags for backward-compatibility.
			UnknownFlags: true,
		},
		RunE: func(c *cobra.Command, args []string) error {
            ...
			sa := istio_agent.NewSDSAgent(proxyConfig.DiscoveryAddress, proxyConfig.ControlPlaneAuthPolicy == meshconfig.AuthenticationPolicy_MUTUAL_TLS,
				pilotCertProvider, jwtPath, outputKeyCertToDir, clusterIDVar.Get())

            ...
			// Start in process SDS.
			_, err = sa.Start(role.Type == model.SidecarProxy, podNamespaceVar.Get())
			if err != nil {
				log.Fatala("Failed to start in-process SDS", err)
			}
    ...

func (sa *SDSAgent) Start(isSidecar bool, podNamespace string) (*sds.Server, error) {
    ...
	workloadSecretCache, _ := sa.newSecretCache(serverOptions)
    ...
	server, err := sds.NewServer(serverOptions, workloadSecretCache, gatewaySecretCache)
    ...
	return server, nil
}

// newSecretCache creates the cache for workload secrets and/or gateway secrets.
func (sa *SDSAgent) newSecretCache(serverOptions sds.Options) (workloadSecretCache *cache.SecretCache, caClient caClientInterface.Client) {
	ret := &secretfetcher.SecretFetcher{}

	// TODO: get the MC public keys from pilot.
	// In node agent, a controller is used getting 'istio-security.istio-system' config map
	// Single caTLSRootCert inside.

	var err error

	// TODO: this should all be packaged in a plugin, possibly with optional compilation.
	log.Infof("serverOptions.CAEndpoint == %v", serverOptions.CAEndpoint)
	if (serverOptions.CAProviderName == "GoogleCA" || strings.Contains(serverOptions.CAEndpoint, "googleapis.com")) &&
		stsclient.GKEClusterURL != "" {
		// Use a plugin to an external CA - this has direct support for the K8S JWT token
		// This is only used if the proper env variables are injected - otherwise the existing Citadel or Istiod will be
		// used.
		caClient, err = gca.NewGoogleCAClient(serverOptions.CAEndpoint, true)
		serverOptions.PluginNames = []string{"GoogleTokenExchange"}
	} else {

		// Determine the default CA.
		// If /etc/certs exists - it means Citadel is used (possibly in a mode to only provision the root-cert, not keys)
		// Otherwise: default to istiod
		//
		// If an explicit CA is configured, assume it is mounting /etc/certs
		var rootCert []byte

		tls := true
		certReadErr := false

		if serverOptions.CAEndpoint == "" {
            ...
		} else {

    - args:
        - proxy
        - sidecar
        - --domain
        ...
      env:
        ...
        - name: PILOT_CERT_PROVIDER
          value: istiod
        - name: CA_ADDR
          value: istiod.istio-system.svc:15012

			// Explicitly configured CA
			log.Infoa("Using user-configured CA ", serverOptions.CAEndpoint)
			if strings.HasSuffix(serverOptions.CAEndpoint, ":15010") {
				log.Warna("Debug mode or IP-secure network")
				tls = false
			} else if serverOptions.TLSEnabled {
				if serverOptions.PilotCertProvider == "istiod" {
					log.Info("istiod uses self-issued certificate")
					if rootCert, err = ioutil.ReadFile(path.Join(CitadelCACertPath, constants.CACertNamespaceConfigMapDataName)); err != nil {
						certReadErr = true
					} else {
						log.Infof("the CA cert of istiod is: %v", string(rootCert))
					}
				} else if serverOptions.PilotCertProvider == "kubernetes" {
					log.Infof("istiod uses the k8s root certificate %v", k8sCAPath)
					if rootCert, err = ioutil.ReadFile(k8sCAPath); err != nil {
						certReadErr = true
					}
				} else if serverOptions.PilotCertProvider == "custom" {
					log.Infof("istiod uses a custom root certificate mounted in a well known location %v",
						cache.DefaultRootCertFilePath)
					if rootCert, err = ioutil.ReadFile(cache.DefaultRootCertFilePath); err != nil {
						certReadErr = true
					}
				} else {
					log.Errorf("unknown cert provider %v", serverOptions.PilotCertProvider)
					certReadErr = true
				}
				if certReadErr {
					rootCert = nil
					log.Fatal("invalid config - port 15012 missing a root certificate")
				}
			}
            ...
		}

	pilotCertProvider = env.RegisterStringVar("PILOT_CERT_PROVIDER", "istiod",
		"the provider of Pilot DNS certificate.").Get()

		sa.RootCert = rootCert
		// Will use TLS unless the reserved 15010 port is used ( istiod on an ipsec/secure VPC)
		// rootCert may be nil - in which case the system roots are used, and the CA is expected to have public key
		// Otherwise assume the injection has mounted /etc/certs/root-cert.pem
		caClient, err = citadel.NewCitadelClient(serverOptions.CAEndpoint, tls, rootCert, serverOptions.ClusterID)
		if err == nil {
			sa.CitadelClient = caClient
		}
	}

	if err != nil {
		log.Errorf("failed to create secretFetcher for workload proxy: %v", err)
		os.Exit(1)
	}
	ret.UseCaClient = true
	ret.CaClient = caClient

	workloadSdsCacheOptions.TrustDomain = serverOptions.TrustDomain
	workloadSdsCacheOptions.Pkcs8Keys = serverOptions.Pkcs8Keys
	workloadSdsCacheOptions.ECCSigAlg = serverOptions.ECCSigAlg
	workloadSdsCacheOptions.Plugins = sds.NewPlugins(serverOptions.PluginNames)
	workloadSdsCacheOptions.OutputKeyCertToDir = serverOptions.OutputKeyCertToDir
	workloadSecretCache = cache.NewSecretCache(ret, sds.NotifyProxy, workloadSdsCacheOptions)
	sa.WorkloadSecrets = workloadSecretCache
	return
}

func (sa *SDSAgent) Start(isSidecar bool, podNamespace string) (*sds.Server, error) {
    ...
	workloadSecretCache, _ := sa.newSecretCache(serverOptions)
    ...
	server, err := sds.NewServer(serverOptions, workloadSecretCache, gatewaySecretCache)
    ...
	return server, nil
}

func NewServer(options Options, workloadSecretCache, gatewaySecretCache cache.SecretManager) (*Server, error) {
	s := &Server{
		workloadSds: newSDSService(workloadSecretCache, false, options.UseLocalJWT,
			options.RecycleInterval, options.JWTPath, options.OutputKeyCertToDir),
		gatewaySds: newSDSService(gatewaySecretCache, true, options.UseLocalJWT,
			options.RecycleInterval, options.JWTPath, options.OutputKeyCertToDir),
	}
	if options.EnableWorkloadSDS {
		if err := s.initWorkloadSdsService(&options); err != nil {
			sdsServiceLog.Errorf("Failed to initialize secret discovery service for workload proxies: %v", err)
			return nil, err
		}
		sdsServiceLog.Infof("SDS gRPC server for workload UDS starts, listening on %q \n", options.WorkloadUDSPath)
	}

	if options.EnableIngressGatewaySDS {
		if err := s.initGatewaySdsService(&options); err != nil {
			sdsServiceLog.Errorf("Failed to initialize secret discovery service for ingress gateway: %v", err)
			return nil, err
		}
		sdsServiceLog.Infof("SDS gRPC server for ingress gateway controller starts, listening on %q \n",
			options.IngressGatewayUDSPath)
	}
    ...
	return s, nil
}

// StreamSecrets serves SDS discovery requests and SDS push requests
func (s *sdsservice) StreamSecrets(stream sds.SecretDiscoveryService_StreamSecretsServer) error {
    ...
	go receiveThread(con, reqChannel, &receiveError)

	for {
		select {
		case discReq, ok := <-reqChannel:
            ...
			secret, err := s.st.GenerateSecret(ctx, conID, resourceName, token)
            ...
			if err := pushSDS(con); err != nil {
				sdsServiceLog.Errorf("%s Close connection. Failed to push key/cert to proxy %q: %v",
					conIDresourceNamePrefix, discReq.Node.Id, err)
				return err
			}
		case <-con.pushChannel:
            ...
			if err := pushSDS(con); err != nil {
				sdsServiceLog.Errorf("%s Close connection. Failed to push key/cert to proxy %q: %v",
					conIDresourceNamePrefix, proxyID, err)
				return err
			}
		}
	}
}

	var node *core.Node
	for {
		// Block until a request is received.
		select {
		case discReq, ok := <-reqChannel:
            ...
		case <-con.pushChannel:
            ...
		}
	}
}

			if con.conID == "" {
				// first request
				if discReq.Node == nil || len(discReq.Node.Id) == 0 {
					sdsServiceLog.Errorf("%s close connection. Missing Node ID in the first request",
						sdsLogPrefix(resourceName))
					return fmt.Errorf("missing Node ID in the first request")
				}
				con.conID = constructConnectionID(discReq.Node.Id)
				con.proxyID = discReq.Node.Id
				con.ResourceName = resourceName
				key := cache.ConnKey{
					ResourceName: resourceName,
					ConnectionID: con.conID,
				}
				addConn(key, con)
				firstRequestFlag = true
				sdsServiceLog.Infof("%s new connection", sdsLogPrefix(resourceName))
			}

func addConn(k cache.ConnKey, conn *sdsConnection) {
	sdsClientsMutex.Lock()
	defer sdsClientsMutex.Unlock()
	conIDresourceNamePrefix := sdsLogPrefix(k.ResourceName)
	sdsServiceLog.Debugf("%s add a new connection", conIDresourceNamePrefix)
	sdsClients[k] = conn
}

			if s.localJWT {
				// Running in-process, no need to pass the token from envoy to agent as in-context - use the file
				tok, err := ioutil.ReadFile(s.jwtPath)
				if err != nil {
					sdsServiceLog.Errorf("Failed to get credential token: %v", err)
					return err
				}
				token = string(tok)
			} else if s.outputKeyCertToDir != "" {
				// Using existing certs and the new SDS - skipToken case is for the old node agent.
			} else if !s.skipToken {
				ctx = stream.Context()
				t, err := getCredentialToken(ctx)
				if err != nil {
					sdsServiceLog.Errorf("%s Close connection. Failed to get credential token from "+
						"incoming request: %v", conIDresourceNamePrefix, err)
					return err
				}
				token = t
			}

			secret, err := s.st.GenerateSecret(ctx, conID, resourceName, token)
            ...
			if err := pushSDS(con); err != nil {
				sdsServiceLog.Errorf("%s Close connection. Failed to push key/cert to proxy %q: %v",
					conIDresourceNamePrefix, discReq.Node.Id, err)
				return err
			}

func (sc *SecretCache) GenerateSecret(ctx context.Context, connectionID, resourceName, token string) (*model.SecretItem, error) {
    ...
	// First try to generate secret from file.
	sdsFromFile, ns, err := sc.generateFileSecret(connKey, token)

	if sdsFromFile {
        ...
		return ns, nil
	}

	if resourceName != RootCertReqResourceName {
		ns, err := sc.generateSecret(ctx, token, connKey, time.Now())
        ...
		return ns, nil
	}

    ...
	rootCert, rootCertExpr := sc.getRootCert()
    ...

	ns = &model.SecretItem{
		ResourceName: resourceName,
		RootCert:     rootCert,
		ExpireTime:   rootCertExpr,
		Token:        token,
		CreatedTime:  t,
		Version:      t.String(),
	}
    ...
	return ns, nil
}

func (sc *SecretCache) generateSecret(ctx context.Context, token string, connKey ConnKey, t time.Time) (*model.SecretItem, error) {
    ...
	exchangedToken, err := sc.getExchangedToken(ctx, token, connKey)

    ...
	options := pkiutil.CertOptions{
		Host:       csrHostName,
		RSAKeySize: keySize,
		PKCS8Key:   sc.configOptions.Pkcs8Keys,
		ECSigAlg:   pkiutil.SupportedECSignatureAlgorithms(sc.configOptions.ECCSigAlg),
	}

	csrPEM, keyPEM, err := pkiutil.GenCSR(options)
    ...

	certChainPEM, err := sc.sendRetriableRequest(ctx, csrPEM, exchangedToken, connKey, true)

	certChain := []byte{}
	for _, c := range certChainPEM {
		certChain = append(certChain, []byte(c)...)
	}

    ...
	rootCert, _ := sc.getRootCert()
	// Leaf cert is element '0'. Root cert is element 'n'.
	rootCertChanged := !bytes.Equal(rootCert, []byte(certChainPEM[length-1]))
	if rootCert == nil || rootCertChanged {
		rootCertExpireTime, err := nodeagentutil.ParseCertAndGetExpiryTimestamp([]byte(certChainPEM[length-1]))
		if err == nil {
			sc.setRootCert([]byte(certChainPEM[length-1]), rootCertExpireTime)
		} else {
			cacheLog.Errorf("%s failed to parse root certificate in CSR response: %v", logPrefix, err)
			rootCertChanged = false
		}
	}

	if rootCertChanged {
		cacheLog.Info("Root cert has changed, start rotating root cert for SDS clients")
		sc.rotate(true /*updateRootFlag*/)
	}

	return &model.SecretItem{
		CertificateChain: certChain,
		PrivateKey:       keyPEM,
		ResourceName:     connKey.ResourceName,
		Token:            token,
		CreatedTime:      t,
		ExpireTime:       expireTime,
		Version:          t.Format("01-02 15:04:05.000"), // Precise enough version based on creation time.
	}, nil
}

	var node *core.Node
	for {
		// Block until a request is received.
		select {
		case discReq, ok := <-reqChannel:
            ...
		case <-con.pushChannel:
			con.mutex.RLock()
			proxyID := con.proxyID
			conID := con.conID
			resourceName := con.ResourceName
			secret := con.secret
			con.mutex.RUnlock()

			if secret == nil {
                ...
			}

			if err := pushSDS(con); err != nil {
				sdsServiceLog.Errorf("%s Close connection. Failed to push key/cert to proxy %q: %v",
					conIDresourceNamePrefix, proxyID, err)
				return err
			}
		}
	}
}

func (sa *SDSAgent) newSecretCache(serverOptions sds.Options) (workloadSecretCache *cache.SecretCache, caClient caClientInterface.Client) {
    ...
	workloadSecretCache = cache.NewSecretCache(ret, sds.NotifyProxy, workloadSdsCacheOptions)
	sa.WorkloadSecrets = workloadSecretCache
	return
}

func NotifyProxy(connKey cache.ConnKey, secret *model.SecretItem) error {
	conIDresourceNamePrefix := sdsLogPrefix(connKey.ResourceName)

	conn := sdsClients[connKey]
    ...
	conn.mutex.Lock()
	conn.secret = secret
	conn.mutex.Unlock()
	sdsClientsMutex.Unlock()

	conn.pushChannel <- &sdsEvent{}
	return nil
}

func (sc *SecretCache) callbackWithTimeout(connKey ConnKey, secret *model.SecretItem) {
	go func() {
        ...
		if err := sc.notifyCallback(connKey, secret); err != nil {
            ...
		}
	}()
	select {
	case <-c:
		return // completed normally
	case <-time.After(notifySecretRetrievalTimeout):
		cacheLog.Warnf("%s notify secret change for proxy got timeout", logPrefix)
	}
}

func (sc *SecretCache) generateFileSecret(connKey ConnKey, token string) (bool, *model.SecretItem, error) {
    ...
	switch {
	// Default root certificate.
	case connKey.ResourceName == RootCertReqResourceName && sc.rootCertificateExist(sc.existingRootCertFile):
		sdsFromFile = true
		ns, err = sc.generateRootCertFromExistingFile(sc.existingRootCertFile, token, connKey)
		sc.addFileWatcher(sc.existingRootCertFile, token, connKey)
	// Default workload certificate.
	case connKey.ResourceName == WorkloadKeyCertResourceName && sc.keyCertificateExist(sc.existingCertChainFile, sc.existingKeyFile):
		sdsFromFile = true
		ns, err = sc.generateKeyCertFromExistingFiles(sc.existingCertChainFile, sc.existingKeyFile, token, connKey)
		// Adding cert is sufficient here as key can't change without changing the cert.
		sc.addFileWatcher(sc.existingCertChainFile, token, connKey)
	default:
		// Check if the resource name refers to a file mounted certificate.
		// Currently used in destination rules and server certs (via metadata).
		// Based on the resource name, we need to read the secret from a file encoded in the resource name.
		cfg, ok := pilotmodel.SdsCertificateConfigFromResourceName(connKey.ResourceName)
		switch {
		case ok && cfg.IsRootCertificate() && sc.rootCertificateExist(cfg.CaCertificatePath):
			sdsFromFile = true
			ns, err = sc.generateRootCertFromExistingFile(cfg.CaCertificatePath, token, connKey)
			sc.addFileWatcher(cfg.CaCertificatePath, token, connKey)
		case ok && cfg.IsKeyCertificate() && sc.keyCertificateExist(cfg.CertificatePath, cfg.PrivateKeyPath):
			sdsFromFile = true
			ns, err = sc.generateKeyCertFromExistingFiles(cfg.CertificatePath, cfg.PrivateKeyPath, token, connKey)
			// Adding cert is sufficient here as key can't change without changing the cert.
			sc.addFileWatcher(cfg.CertificatePath, token, connKey)
		}
	}
    ...
	return sdsFromFile, nil, nil
}

func (sc *SecretCache) addFileWatcher(file string, token string, connKey ConnKey) {
    ...
	go func() {
		var timerC <-chan time.Time
		for {
			select {
			case <-timerC:
                ...
				connKeys := sc.fileCerts[npath]
                ...
				for ckey := range connKeys {
					if _, ok := sc.secrets.Load(ckey); ok {
						// Regenerate the Secret and trigger the callback that pushes the secrets to proxy.
						if _, secret, err := sc.generateFileSecret(ckey, token); err != nil {
							cacheLog.Errorf("%v: error in generating secret after file change [%s] %v", ckey, file, err)
						} else {
							cacheLog.Infof("%v: file changed, triggering secret push to proxy [%s]", ckey, file)
							sc.callbackWithTimeout(ckey, secret)
						}
					}
				}
			case e := <-sc.certWatcher.Events(file):
				if len(e.Op.String()) > 0 { // To avoid spurious events, mainly coming from tests.
					// Use a timer to debounce watch updates
					if timerC == nil {
						timerC = time.After(100 * time.Millisecond) // TODO: Make this configurable if needed.
					}
				}
			}
		}
	}()
}

func (sc *SecretCache) generateSecret(ctx context.Context, token string, connKey ConnKey, t time.Time) (*model.SecretItem, error) {
    ...
	if rootCertChanged {
		cacheLog.Info("Root cert has changed, start rotating root cert for SDS clients")
		sc.rotate(true /*updateRootFlag*/)
	}

	return &model.SecretItem{
		CertificateChain: certChain,
		PrivateKey:       keyPEM,
		ResourceName:     connKey.ResourceName,
		Token:            token,
		CreatedTime:      t,
		ExpireTime:       expireTime,
		Version:          t.Format("01-02 15:04:05.000"), // Precise enough version based on creation time.
	}, nil
}

func (sc *SecretCache) rotate(updateRootFlag bool) {
    ...
	sc.secrets.Range(func(k interface{}, v interface{}) bool {
        ...
		if updateRootFlag {
            ...
			sc.callbackWithTimeout(connKey, ns)

			return true
		}

        ...
		// Re-generate secret if it's expired.
		if sc.shouldRotate(&secret) {
			atomic.AddUint64(&sc.secretChangedCount, 1)
			// Send the notification to close the stream if token is expired, so that client could re-connect with a new token.
			if sc.isTokenExpired(&secret) {
                ...
				sc.callbackWithTimeout(connKey, nil /*nil indicates close the streaming connection to proxy*/)
				return true
			}

			wg.Add(1)
			go func() {
                ...
				ns, err := sc.generateSecret(context.Background(), secret.Token, connKey, now)
				if err != nil {
					cacheLog.Errorf("%s failed to rotate secret: %v", logPrefix, err)
					return
				}
                ...
				sc.callbackWithTimeout(connKey, ns)

			}()
		}

		return true
	})
    ...
}