Store and return rpc error in acl cache entries

This commit is contained in:
Kyle Havlovitz 2022-04-28 09:08:55 -07:00
parent e4008124bd
commit f84ed5f70b
4 changed files with 42 additions and 12 deletions

View File

@ -360,20 +360,20 @@ func (r *ACLResolver) fetchAndCacheIdentityFromToken(token string, cached *struc
err := r.backend.RPC("ACL.TokenRead", &req, &resp)
if err == nil {
if resp.Token == nil {
r.cache.PutIdentity(cacheID, nil)
r.cache.PutIdentity(cacheID, nil, nil)
return nil, acl.ErrNotFound
} else if resp.Token.Local && r.config.Datacenter != resp.SourceDatacenter {
r.cache.PutIdentity(cacheID, nil)
r.cache.PutIdentity(cacheID, nil, nil)
return nil, acl.PermissionDeniedError{Cause: fmt.Sprintf("This is a local token in datacenter %q", resp.SourceDatacenter)}
} else {
r.cache.PutIdentity(cacheID, resp.Token)
r.cache.PutIdentity(cacheID, resp.Token, nil)
return resp.Token, nil
}
}
if acl.IsErrNotFound(err) {
// Make sure to remove from the cache if it was deleted
r.cache.PutIdentity(cacheID, nil)
r.cache.PutIdentity(cacheID, nil, err)
return nil, acl.ErrNotFound
}
@ -381,11 +381,11 @@ func (r *ACLResolver) fetchAndCacheIdentityFromToken(token string, cached *struc
// some other RPC error
if cached != nil && (r.config.ACLDownPolicy == "extend-cache" || r.config.ACLDownPolicy == "async-cache") {
// extend the cache
r.cache.PutIdentity(cacheID, cached.Identity)
r.cache.PutIdentity(cacheID, cached.Identity, err)
return cached.Identity, nil
}
r.cache.PutIdentity(cacheID, nil)
r.cache.PutIdentity(cacheID, nil, err)
return nil, err
}
@ -402,7 +402,7 @@ func (r *ACLResolver) resolveIdentityFromToken(token string) (structs.ACLIdentit
cacheEntry := r.cache.GetIdentity(tokenSecretCacheID(token))
if cacheEntry != nil && cacheEntry.Age() <= r.config.ACLTokenTTL {
metrics.IncrCounter([]string{"acl", "token", "cache_hit"}, 1)
return cacheEntry.Identity, nil
return cacheEntry.Identity, cacheEntry.Error
}
metrics.IncrCounter([]string{"acl", "token", "cache_miss"}, 1)
@ -416,7 +416,7 @@ func (r *ACLResolver) resolveIdentityFromToken(token string) (structs.ACLIdentit
waitForResult := cacheEntry == nil || r.config.ACLDownPolicy != "async-cache"
if !waitForResult {
// waitForResult being false requires the cacheEntry to not be nil
return cacheEntry.Identity, nil
return cacheEntry.Identity, cacheEntry.Error
}
// block on the read here, this is why we don't need chan buffering
@ -549,7 +549,7 @@ func (r *ACLResolver) maybeHandleIdentityErrorDuringFetch(identity structs.ACLId
if acl.IsErrNotFound(err) {
// make sure to indicate that this identity is no longer valid within
// the cache
r.cache.PutIdentity(tokenSecretCacheID(identity.SecretToken()), nil)
r.cache.PutIdentity(tokenSecretCacheID(identity.SecretToken()), nil, err)
// Do not touch the cache. Getting a top level ACL not found error
// only indicates that the secret token used in the request

View File

@ -1264,6 +1264,35 @@ func TestACLResolver_DownPolicy(t *testing.T) {
requireIdentityCached(t, r, tokenSecretCacheID("found"), false, "no longer cached")
})
t.Run("Cache-Error", func(t *testing.T) {
_, rawToken, _ := testIdentityForToken("found")
foundToken := rawToken.(*structs.ACLToken)
secretID := foundToken.SecretID
delegate := &ACLResolverTestDelegate{
enabled: true,
datacenter: "dc1",
legacy: false,
localTokens: false,
localPolicies: false,
tokenReadFn: func(_ *structs.ACLTokenGetRequest, reply *structs.ACLTokenResponse) error {
return acl.ErrPermissionDenied
},
}
r := newTestACLResolver(t, delegate, func(config *ACLResolverConfig) {
config.Config.ACLDownPolicy = "extend-cache"
})
// Attempt to resolve the token - this should set up the cache entry with a nil
// identity and permission denied error.
authz, err := r.ResolveToken(secretID)
require.NoError(t, err)
require.NotNil(t, authz)
entry := r.cache.GetIdentity(tokenSecretCacheID(secretID))
require.Nil(t, entry.Identity)
require.Equal(t, acl.ErrPermissionDenied, entry.Error)
})
t.Run("PolicyResolve-TokenNotFound", func(t *testing.T) {
_, rawToken, _ := testIdentityForToken("found")
foundToken := rawToken.(*structs.ACLToken)

View File

@ -27,6 +27,7 @@ type ACLCaches struct {
type IdentityCacheEntry struct {
Identity ACLIdentity
CacheTime time.Time
Error error
}
func (e *IdentityCacheEntry) Age() time.Duration {
@ -187,12 +188,12 @@ func (c *ACLCaches) GetRole(roleID string) *RoleCacheEntry {
}
// PutIdentity adds a new identity to the cache
func (c *ACLCaches) PutIdentity(id string, ident ACLIdentity) {
func (c *ACLCaches) PutIdentity(id string, ident ACLIdentity, err error) {
if c == nil || c.identities == nil {
return
}
c.identities.Add(id, &IdentityCacheEntry{Identity: ident, CacheTime: time.Now()})
c.identities.Add(id, &IdentityCacheEntry{Identity: ident, CacheTime: time.Now(), Error: err})
}
func (c *ACLCaches) PutPolicy(policyId string, policy *ACLPolicy) {

View File

@ -47,7 +47,7 @@ func TestStructs_ACLCaches(t *testing.T) {
require.NoError(t, err)
require.NotNil(t, cache)
cache.PutIdentity("foo", &ACLToken{})
cache.PutIdentity("foo", &ACLToken{}, nil)
entry := cache.GetIdentity("foo")
require.NotNil(t, entry)
require.NotNil(t, entry.Identity)