Create Vault client once then renew secret token as needed

* SecretStoreClient created exactly once as part of first reconciliation
* Vault token owned by SecretStoreClient is continually renewed in a
  background thread that manages its lifecycle
* Lifecycle management thread is cognizant of Vault token's max TTL
  (defaults to 32 days) and will force a fresh Vault login when reached

Author: George Harley

internal/cluster_reference.go

@@ -27,7 +27,7 @@ func (c ClusterCredentials) Data(key string) ([]byte, bool) {
 	return result, ok
 }
 
-var SecretStoreClientInitializer = InitializeSecretStoreClient
+var SecretStoreClientProvider = GetSecretStoreClient
 
 var (
 	NoSuchRabbitmqClusterError = errors.New("RabbitmqCluster object does not exist")
@@ -64,8 +64,8 @@ func ParseRabbitmqClusterReference(ctx context.Context, c client.Client, rmq top
 	var credentialsProvider CredentialsProvider
 	svc := &corev1.Service{}
 	if cluster.Spec.SecretBackend.Vault != nil && cluster.Spec.SecretBackend.Vault.DefaultUserPath != "" {
-		// ask the configured secure store for the credentials available at the path retrived from the cluster resource
-		secretStoreClient, err := SecretStoreClientInitializer(cluster.Spec.SecretBackend.Vault)
+		// ask the configured secure store for the credentials available at the path retrieved from the cluster resource
+		secretStoreClient, err := SecretStoreClientProvider(cluster.Spec.SecretBackend.Vault)
 		if err != nil {
 			return nil, nil, nil, fmt.Errorf("unable to create a client connection to secret store: %w", err)
 		}
@@ -74,6 +74,7 @@ func ParseRabbitmqClusterReference(ctx context.Context, c client.Client, rmq top
 		if err != nil {
 			return nil, nil, nil, fmt.Errorf("unable to retrieve credentials from secret store: %w", err)
 		}
+
 		credentialsProvider = credsProv
 
 		if err := c.Get(ctx, types.NamespacedName{Namespace: namespace, Name: cluster.ObjectMeta.Name}, svc); err != nil {

internal/cluster_reference_test.go

@@ -156,13 +156,13 @@ var _ = Describe("ParseRabbitmqClusterReference", func() {
 
 				fakeSecretStoreClient = &internalfakes.FakeSecretStoreClient{}
 				fakeSecretStoreClient.ReadCredentialsReturns(fakeCredentialsProvider, nil)
-				internal.SecretStoreClientInitializer = func(vaultSpec *rabbitmqv1beta1.VaultSpec) (internal.SecretStoreClient, error) {
+				internal.SecretStoreClientProvider = func(vaultSpec *rabbitmqv1beta1.VaultSpec) (internal.SecretStoreClient, error) {
 					return fakeSecretStoreClient, nil
 				}
 			})
 
 			AfterEach(func() {
-				internal.SecretStoreClientInitializer = internal.InitializeSecretStoreClient
+				internal.SecretStoreClientProvider = internal.GetSecretStoreClient
 			})
 
 			JustBeforeEach(func() {

internal/vault_reader.go

@@ -4,9 +4,12 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"sync"
 
 	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/api/v1beta1"
 
+	ctrl "sigs.k8s.io/controller-runtime"
+
 	vault "github.com/hashicorp/vault/api"
 )
 
@@ -38,9 +41,42 @@ type VaultClient struct {
 	Reader SecretReader
 }
 
-var ServiceAccountTokenReader = ReadServiceAccountToken
-var VaultClientTokenReader = ReadVaultClientToken
-var VaultAuthenticator = LoginToVault
+var ReadServiceAccountTokenFunc = ReadServiceAccountToken
+var ReadVaultClientSecretFunc = ReadVaultClientSecret
+var LoginToVaultFunc = LoginToVault
+
+var createSecretStoreClientOnce sync.Once
+var SecretClient SecretStoreClient
+var SecretClientCreationError error
+
+func GetSecretStoreClient(vaultSpec *rabbitmqv1beta1.VaultSpec) (SecretStoreClient, error) {
+	createSecretStoreClientOnce.Do(InitializeClient(vaultSpec))
+	return SecretClient, SecretClientCreationError
+}
+
+func InitializeClient(vaultSpec *rabbitmqv1beta1.VaultSpec) func() {
+	return func() {
+		// VAULT_ADDR environment variable will be the address that pod uses to communicate with Vault.
+		config := vault.DefaultConfig() // modify for more granular configuration
+		vaultClient, err := vault.NewClient(config)
+		if err != nil {
+			SecretClientCreationError = fmt.Errorf("unable to initialize Vault client: %w", err)
+			return
+		}
+
+		_, err = login(vaultClient, vaultSpec)
+		if err != nil {
+			SecretClientCreationError = fmt.Errorf("unable to login to Vault: %w", err)
+			return
+		}
+
+		SecretClient = VaultClient{
+			Reader: &VaultSecretReader{client: vaultClient},
+		}
+
+		go renewToken(vaultClient, vaultSpec)
+	}
+}
 
 func (vc VaultClient) ReadCredentials(path string) (CredentialsProvider, error) {
 	secret, err := vc.Reader.ReadSecret(path)
@@ -110,46 +146,105 @@ func availableKeys(m map[string]interface{}) []string {
 	return result
 }
 
-func InitializeSecretStoreClient(vaultSpec *rabbitmqv1beta1.VaultSpec) (SecretStoreClient, error) {
+func login(vaultClient *vault.Client, vaultSpec *rabbitmqv1beta1.VaultSpec) (*vault.Secret, error) {
+	logger := ctrl.LoggerFrom(nil)
+
 	// GCH TODO return to this...
 	// role := vaultSpec.Role
 	// if role == "" {
 	// 	return nil, errors.New("no role value set in Vault secret backend")
 	// }
 	role := "messaging-topology-operator"
 
-	// For now, the VAULT_ADDR environment variable will be the address that your pod uses to communicate with Vault.
-	config := vault.DefaultConfig() // modify for more granular configuration
-
-	vaultClient, err := vault.NewClient(config)
-	if err != nil {
-		return nil, fmt.Errorf("unable to initialize Vault client: %w", err)
-	}
-
 	var annotations = vaultSpec.Annotations
 	if annotations["vault.hashicorp.com/namespace"] != "" {
 		vaultClient.SetNamespace(annotations["vault.hashicorp.com/namespace"])
 	}
 
-	jwt, err := ServiceAccountTokenReader()
+	jwt, err := ReadServiceAccountTokenFunc()
 	if err != nil {
 		return nil, fmt.Errorf("unable to read file containing service account token: %w", err)
 	}
 
 	loginAuthPath := defaultAuthPath
+	annotations = vaultSpec.Annotations
 	if annotations["vault.hashicorp.com/auth-path"] != "" {
 		loginAuthPath = annotations["vault.hashicorp.com/auth-path"]
 	}
 
-	vaultToken, err := VaultClientTokenReader(vaultClient, string(jwt), role, loginAuthPath)
+	logger.Info("Authenticating to Vault")
+
+	vaultSecret, err := ReadVaultClientSecretFunc(vaultClient, string(jwt), role, loginAuthPath)
 	if err != nil {
-		return nil, fmt.Errorf("unable to read Vault client token: %w", err)
+		return nil, fmt.Errorf("unable to obtain Vault client secret: %w", err)
+	}
+
+	if vaultSecret == nil || vaultSecret.Auth == nil || vaultSecret.Auth.ClientToken == "" {
+		return nil, fmt.Errorf("no client token found in Vault secret")
+	}
+
+	vaultClient.SetToken(vaultSecret.Auth.ClientToken)
+	return vaultSecret, nil
+}
+
+func renewToken(client *vault.Client, vaultSpec *rabbitmqv1beta1.VaultSpec) {
+	logger := ctrl.LoggerFrom(nil)
+
+	for {
+		vaultLoginResp, err := login(client, vaultSpec)
+		if err != nil {
+			logger.Error(err, "unable to authenticate to Vault server")
+		}
+
+		err = manageTokenLifecycle(client, vaultLoginResp)
+		if err != nil {
+			logger.Error(err, "unable to start managing the Vault token lifecycle")
+		}
 	}
+}
 
-	// use the Vault token for making all future calls to Vault
-	vaultClient.SetToken(vaultToken)
+func manageTokenLifecycle(client *vault.Client, token *vault.Secret) error {
+	logger := ctrl.LoggerFrom(nil)
+
+	if token == nil || token.Auth == nil {
+		logger.Info("No Vault secret available. Re-attempting login")
+		return nil
+	}
+
+	renew := token.Auth.Renewable
+	if !renew {
+		logger.Info("Token is not configured to be renewable. Re-attempting login")
+		return nil
+	}
 
-	return VaultClient{Reader: &VaultSecretReader{client: vaultClient}}, nil
+	watcher, err := client.NewLifetimeWatcher(&vault.LifetimeWatcherInput{
+		Secret: token,
+	})
+	if err != nil {
+		return fmt.Errorf("unable to initialize new lifetime watcher for renewing auth token: %w", err)
+	}
+
+	go watcher.Start()
+	defer watcher.Stop()
+
+	for {
+		select {
+		// `DoneCh` will return if renewal fails, or if the remaining lease duration is
+		// under a built-in threshold and either renewing is not extending it or
+		// renewing is disabled.  In any case, the caller needs to attempt to log in again.
+		case err := <-watcher.DoneCh():
+			if err != nil {
+				logger.Error(err, "Failed to renew Vault token. Re-attempting login")
+				return nil
+			}
+			logger.Info("Token can no longer be renewed. Re-attempting login.")
+			return nil
+
+		// Successfully completed renewal
+		case renewal := <-watcher.RenewCh():
+			logger.Info("Successfully renewed Vault token", "renewal info", renewal)
+		}
+	}
 }
 
 func ReadServiceAccountToken() ([]byte, error) {
@@ -164,23 +259,13 @@ func ReadServiceAccountToken() ([]byte, error) {
 	return token, nil
 }
 
-func ReadVaultClientToken(vaultClient *vault.Client, jwtToken string, vaultRole string, authPath string) (string, error) {
+func ReadVaultClientSecret(vaultClient *vault.Client, jwtToken string, vaultRole string, authPath string) (*vault.Secret, error) {
 	params := map[string]interface{}{
 		"jwt":  jwtToken,
 		"role": vaultRole, // the name of the role in Vault that was created with this app's Kubernetes service account bound to it
 	}
 
-	// log in to Vault's Kubernetes auth method
-	resp, err := VaultAuthenticator(vaultClient, authPath, params)
-	if err != nil {
-		return "", fmt.Errorf("unable to log in with Kubernetes auth: %w", err)
-	}
-
-	// return the Vault client token provided in the login response
-	if resp == nil || resp.Auth == nil || resp.Auth.ClientToken == "" {
-		return "", fmt.Errorf("no client token found in Vault login response")
-	}
-	return resp.Auth.ClientToken, nil
+	return LoginToVaultFunc(vaultClient, authPath, params)
 }
 
 func LoginToVault(vaultClient *vault.Client, authPath string, params map[string]interface{}) (*vault.Secret, error) {

internal/vault_reader_test.go

@@ -290,15 +290,21 @@ var _ = Describe("VaultReader", func() {
 	Describe("Initialize secret store client", func() {
 		var (
 			vaultSpec *rabbitmqv1beta1.VaultSpec
+			getSecretStoreClientTester func(vaultSpec *rabbitmqv1beta1.VaultSpec)(internal.SecretStoreClient, error)
 		)
 
 		When("vault spec has no role value", func() {
 			BeforeEach(func() {
 				vaultSpec = &rabbitmqv1beta1.VaultSpec{}
+
+				getSecretStoreClientTester = func(vaultSpec *rabbitmqv1beta1.VaultSpec) (internal.SecretStoreClient, error) {
+					internal.InitializeClient(vaultSpec)()
+					return internal.SecretClient, internal.SecretClientCreationError
+				}
 			})
 
 			JustBeforeEach(func() {
-				secretStoreClient, err = internal.InitializeSecretStoreClient(vaultSpec)
+				secretStoreClient, err = getSecretStoreClientTester(vaultSpec)
 			})
 
 			PIt("should return a nil secret store client", func() {
@@ -313,13 +319,19 @@ var _ = Describe("VaultReader", func() {
 
 		When("service account token is not in the expected place", func() {
 			BeforeEach(func() {
+				internal.SecretClient = nil
+				internal.SecretClientCreationError = nil
 				vaultSpec = &rabbitmqv1beta1.VaultSpec{
 					Role: "cheese-and-ham",
 				}
+				getSecretStoreClientTester = func(vaultSpec *rabbitmqv1beta1.VaultSpec) (internal.SecretStoreClient, error) {
+					internal.InitializeClient(vaultSpec)()
+					return internal.SecretClient, internal.SecretClientCreationError
+				}
 			})
 
 			JustBeforeEach(func() {
-				secretStoreClient, err = internal.InitializeSecretStoreClient(vaultSpec)
+				secretStoreClient, err = getSecretStoreClientTester(vaultSpec)
 			})
 
 			It("should return a nil secret store client", func() {
@@ -332,26 +344,32 @@ var _ = Describe("VaultReader", func() {
 			})
 		})
 
-		When("unable to log into vault to obtain client token", func() {
+		When("unable to log into vault to obtain client secret", func() {
 			BeforeEach(func() {
+				internal.SecretClient = nil
+				internal.SecretClientCreationError = nil
 				vaultSpec = &rabbitmqv1beta1.VaultSpec{
 					Role: "cheese-and-ham",
 				}
-				internal.ServiceAccountTokenReader = func() ([]byte, error) {
+				internal.ReadServiceAccountTokenFunc = func() ([]byte, error) {
 					return []byte("token"), nil
 				}
-				internal.VaultAuthenticator = func(vaultClient *vault.Client, authPath string, params map[string]interface{}) (*vault.Secret, error) {
+				internal.LoginToVaultFunc = func(vaultClient *vault.Client, authPath string, params map[string]interface{}) (*vault.Secret, error) {
 					return nil, errors.New("login failed (quickly!)")
 				}
+				getSecretStoreClientTester = func(vaultSpec *rabbitmqv1beta1.VaultSpec) (internal.SecretStoreClient, error) {
+					internal.InitializeClient(vaultSpec)()
+					return internal.SecretClient, internal.SecretClientCreationError
+				}
 			})
 
 			AfterEach(func() {
-				internal.ServiceAccountTokenReader = internal.ReadServiceAccountToken
-				internal.VaultAuthenticator = internal.LoginToVault
+				internal.ReadServiceAccountTokenFunc = internal.ReadServiceAccountToken
+				internal.LoginToVaultFunc = internal.LoginToVault
 			})
 
 			JustBeforeEach(func() {
-				secretStoreClient, err = internal.InitializeSecretStoreClient(vaultSpec)
+				secretStoreClient, err = getSecretStoreClientTester(vaultSpec)
 			})
 
 			It("should return a nil secret store client", func() {
@@ -360,30 +378,48 @@ var _ = Describe("VaultReader", func() {
 
 			It("should have returned an error", func() {
 				Expect(err).To(HaveOccurred())
-				Expect(err.Error()).To(ContainSubstring("unable to log in with Kubernetes"))
+				Expect(err.Error()).To(ContainSubstring("unable to obtain Vault client secret"))
 			})
 		})
 
-		When("client token obtained from vault", func() {
+		When("client secret obtained from vault", func() {
 			BeforeEach(func() {
+				internal.SecretClient = nil
+				internal.SecretClientCreationError = nil
 				vaultSpec = &rabbitmqv1beta1.VaultSpec{
 					Role: "cheese-and-ham",
 				}
-				internal.ServiceAccountTokenReader = func() ([]byte, error) {
+				internal.ReadServiceAccountTokenFunc = func() ([]byte, error) {
 					return []byte("token"), nil
 				}
-				internal.VaultClientTokenReader = func(vaultClient *vault.Client, jwtToken string, vaultRole string, authPath string) (string, error) {
-					return "vault-token", nil
+				internal.LoginToVaultFunc = func(vaultClient *vault.Client, authPath string, params map[string]interface{}) (*vault.Secret, error) {
+					return &vault.Secret{
+						Auth: &vault.SecretAuth{
+							ClientToken: "vault-secret-token",
+						},
+					}, nil
+				}
+				internal.ReadVaultClientSecretFunc = func(vaultClient *vault.Client, jwtToken string, vaultRole string, authPath string) (*vault.Secret, error) {
+					return &vault.Secret{
+						Auth: &vault.SecretAuth{
+							ClientToken: "vault-secret-token",
+						},
+					}, nil
+				}
+				getSecretStoreClientTester = func(vaultSpec *rabbitmqv1beta1.VaultSpec) (internal.SecretStoreClient, error) {
+					internal.InitializeClient(vaultSpec)()
+					return internal.SecretClient, internal.SecretClientCreationError
 				}
 			})
 
 			AfterEach(func() {
-				internal.ServiceAccountTokenReader = internal.ReadServiceAccountToken
-				internal.VaultClientTokenReader = internal.ReadVaultClientToken
+				internal.ReadServiceAccountTokenFunc = internal.ReadServiceAccountToken
+				internal.LoginToVaultFunc = internal.LoginToVault
+				internal.ReadVaultClientSecretFunc = internal.ReadVaultClientSecret
 			})
 
 			JustBeforeEach(func() {
-				secretStoreClient, err = internal.InitializeSecretStoreClient(vaultSpec)
+				secretStoreClient, err = getSecretStoreClientTester(vaultSpec)
 			})
 
 			It("should not error", func() {