Improve trusted cert loading in Certificate Auth (#27902)

* Improve trusted cert loading in Certificate Auth

Currently, cert auth has a cache of certName->trusted certificate data.  This cache is updated lazily on login.  In highly concurrent situations, several logins
of the same cert or more likely, logins not specifying role name may happen simulataneously.  In the status quo, each results in going to storage, fetching the role data
(or all roles!), unmarshalling, and certificate parsing.

This change puts a lock matrix in front of the cache miss scenario, so only one of the logins will load and process the role data.  In addition, we treat
the absent role name specially, caching it separately so that it cannot be flushed by eviction on the role cache.

* changelog

* cleanup

builtin/credential/cert/backend.go

@@ -18,6 +18,7 @@ import (
 	"github.com/hashicorp/go-multierror"
 	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/hashicorp/vault/sdk/framework"
+	"github.com/hashicorp/vault/sdk/helper/locksutil"
 	"github.com/hashicorp/vault/sdk/helper/ocsp"
 	"github.com/hashicorp/vault/sdk/logical"
 )
@@ -43,7 +44,8 @@ func Backend() *backend {
 	// ignoring the error as it only can occur with <= 0 size
 	cache, _ := lru.New[string, *trusted](defaultRoleCacheSize)
 	b := backend{
-		trustedCache: cache,
+		trustedCache:      cache,
+		trustedCacheLocks: locksutil.CreateLocks(),
 	}
 	b.Backend = &framework.Backend{
 		Help: backendHelp,
@@ -90,6 +92,8 @@ type backend struct {
 
 	trustedCache         *lru.Cache[string, *trusted]
 	trustedCacheDisabled atomic.Bool
+	trustedCacheLocks    []*locksutil.LockEntry
+	trustedCacheFull     atomic.Pointer[trusted]
 }
 
 func (b *backend) initialize(ctx context.Context, req *logical.InitializationRequest) error {
@@ -137,7 +141,7 @@ func (b *backend) updatedConfig(config *config) {
 	case config.RoleCacheSize < 0:
 		// Just to clean up memory
 		b.trustedCacheDisabled.Store(true)
-		b.trustedCache.Purge()
+		b.flushTrustedCache()
 	case config.RoleCacheSize == 0:
 		config.RoleCacheSize = defaultRoleCacheSize
 		fallthrough
@@ -200,6 +204,7 @@ func (b *backend) flushTrustedCache() {
 	if b.trustedCache != nil { // defensive
 		b.trustedCache.Purge()
 	}
+	b.trustedCacheFull.Store(nil)
 }
 
 const backendHelp = `

builtin/credential/cert/path_login.go

@@ -17,6 +17,8 @@ import (
 	"net/url"
 	"strings"
 
+	"github.com/hashicorp/vault/sdk/helper/locksutil"
+
 	"github.com/hashicorp/errwrap"
 	"github.com/hashicorp/go-multierror"
 	"github.com/hashicorp/vault/sdk/framework"
@@ -596,15 +598,39 @@ func (b *backend) certificateExtensionsMetadata(clientCert *x509.Certificate, co
 
 func (b *backend) getTrustedCerts(ctx context.Context, storage logical.Storage, certName string) (pool *x509.CertPool, trusted []*ParsedCert, trustedNonCAs []*ParsedCert, conf *ocsp.VerifyConfig) {
 	if !b.trustedCacheDisabled.Load() {
-		if trusted, found := b.trustedCache.Get(certName); found {
+		trusted, found := b.getTrustedCertsFromCache(certName)
+		if found {
 			return trusted.pool, trusted.trusted, trusted.trustedNonCAs, trusted.ocspConf
 		}
 	}
 	return b.loadTrustedCerts(ctx, storage, certName)
 }
 
+func (b *backend) getTrustedCertsFromCache(certName string) (*trusted, bool) {
+	if certName == "" {
+		trusted := b.trustedCacheFull.Load()
+		if trusted != nil {
+			return trusted, true
+		}
+	} else if trusted, found := b.trustedCache.Get(certName); found {
+		return trusted, true
+	}
+	return nil, false
+}
+
 // loadTrustedCerts is used to load all the trusted certificates from the backend
 func (b *backend) loadTrustedCerts(ctx context.Context, storage logical.Storage, certName string) (pool *x509.CertPool, trustedCerts []*ParsedCert, trustedNonCAs []*ParsedCert, conf *ocsp.VerifyConfig) {
+	lock := locksutil.LockForKey(b.trustedCacheLocks, certName)
+	lock.Lock()
+	defer lock.Unlock()
+
+	if !b.trustedCacheDisabled.Load() {
+		trusted, found := b.getTrustedCertsFromCache(certName)
+		if found {
+			return trusted.pool, trusted.trusted, trusted.trustedNonCAs, trusted.ocspConf
+		}
+	}
+
 	pool = x509.NewCertPool()
 	trustedCerts = make([]*ParsedCert, 0)
 	trustedNonCAs = make([]*ParsedCert, 0)
@@ -672,12 +698,17 @@ func (b *backend) loadTrustedCerts(ctx context.Context, storage logical.Storage,
 	}
 
 	if !b.trustedCacheDisabled.Load() {
-		b.trustedCache.Add(certName, &trusted{
+		entry := &trusted{
 			pool:          pool,
 			trusted:       trustedCerts,
 			trustedNonCAs: trustedNonCAs,
 			ocspConf:      conf,
-		})
+		}
+		if certName == "" {
+			b.trustedCacheFull.Store(entry)
+		} else {
+			b.trustedCache.Add(certName, entry)
+		}
 	}
 	return
 }

changelog/27902.txt

@@ -0,0 +1,5 @@
+```release-note:improvement
+auth/cert: Cache full list of role trust information separately to avoid
+eviction, and avoid duplicate loading during multiple simultaneous logins on
+the same role.
+```