client: adding token resolution logic

client/acl.go

@@ -0,0 +1,165 @@
+package client
+
+import (
+	"time"
+
+	metrics "github.com/armon/go-metrics"
+	"github.com/hashicorp/nomad/acl"
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+// cachedACLValue is used to manage ACL Token or Policy TTLs
+type cachedACLValue struct {
+	Token     *structs.ACLToken
+	Policy    *structs.ACLPolicy
+	CacheTime time.Time
+}
+
+// Age is the time since the token was cached
+func (c *cachedACLValue) Age() time.Duration {
+	return time.Now().Sub(c.CacheTime)
+}
+
+// resolveToken is used to translate an ACL Token Secret ID into
+// an ACL object, nil if ACLs are disabled, or an error.
+func (c *Client) resolveToken(secretID string) (*acl.ACL, error) {
+	// Fast-path if ACLs are disabled
+	if !c.config.ACLEnabled {
+		return nil, nil
+	}
+	defer metrics.MeasureSince([]string{"client", "acl", "resolveToken"}, time.Now())
+
+	// Resolve the token value
+	token, err := c.resolveTokenValue(secretID)
+	if err != nil {
+		return nil, err
+	}
+	if token == nil {
+		return nil, structs.TokenNotFound
+	}
+
+	// Check if this is a management token
+	if token.Type == structs.ACLManagementToken {
+		return acl.ManagementACL, nil
+	}
+
+	// Resolve the policies
+	policies, err := c.resolvePolicies(token.Policies)
+	if err != nil {
+		return nil, err
+	}
+
+	// Resolve the ACL object
+	aclObj, err := structs.CompileACLObject(c.aclCache, policies)
+	if err != nil {
+		return nil, err
+	}
+	return aclObj, nil
+}
+
+// resolveTokenValue is used to translate a secret ID into an ACL token with caching
+// We use a local cache up to the TTL limit, and then resolve via a server. If we cannot
+// reach a server, but have a cached value we extend the TTL to gracefully handle outages.
+func (c *Client) resolveTokenValue(secretID string) (*structs.ACLToken, error) {
+	// Hot-path the anonymous token
+	if secretID == "" {
+		return structs.AnonymousACLToken, nil
+	}
+
+	// Lookup the token in the cache
+	raw, ok := c.tokenCache.Get(secretID)
+	if ok {
+		cached := raw.(*cachedACLValue)
+		if cached.Age() <= c.config.ACLTokenTTL {
+			return cached.Token, nil
+		}
+	}
+
+	// Lookup the token
+	req := structs.ResolveACLTokenRequest{
+		SecretID:     secretID,
+		QueryOptions: structs.QueryOptions{Region: c.Region()},
+	}
+	var resp structs.ResolveACLTokenResponse
+	if err := c.RPC("ACL.ResolveToken", &req, &resp); err != nil {
+		// If we encounter an error but have a cached value, mask the error and extend the cache
+		if ok {
+			c.logger.Printf("[WARN] client: failed to resolve token, using expired cached value: %v", err)
+			cached := raw.(*cachedACLValue)
+			return cached.Token, nil
+		}
+		return nil, err
+	}
+
+	// Cache the response (positive or negative)
+	c.tokenCache.Add(secretID, &cachedACLValue{
+		Token:     resp.Token,
+		CacheTime: time.Now(),
+	})
+	return resp.Token, nil
+}
+
+// resolvePolicies is used to translate a set of named ACL policies into the objects.
+// We cache the policies locally, and fault them from a server as necessary. Policies
+// are cached for a TTL, and then refreshed. If a server cannot be reached, the cache TTL
+// will be ignored to gracefully handle outages.
+func (c *Client) resolvePolicies(policies []string) ([]*structs.ACLPolicy, error) {
+	var out []*structs.ACLPolicy
+	var expired []*structs.ACLPolicy
+	var missing []string
+
+	// Scan the cache for each policy
+	for _, policyName := range policies {
+		// Lookup the policy in the cache
+		raw, ok := c.policyCache.Get(policyName)
+		if !ok {
+			missing = append(missing, policyName)
+			continue
+		}
+
+		// Check if the cached value is valid or expired
+		cached := raw.(*cachedACLValue)
+		if cached.Age() <= c.config.ACLPolicyTTL {
+			out = append(out, cached.Policy)
+		} else {
+			expired = append(expired, cached.Policy)
+		}
+	}
+
+	// Hot-path if we have no missing or expired policies
+	if len(missing)+len(expired) == 0 {
+		return out, nil
+	}
+
+	// Lookup the missing and expired policies
+	fetch := missing
+	for _, p := range expired {
+		fetch = append(fetch, p.Name)
+	}
+	req := structs.ACLPolicySetRequest{
+		Names:        fetch,
+		QueryOptions: structs.QueryOptions{Region: c.Region()},
+	}
+	var resp structs.ACLPolicySetResponse
+	if err := c.RPC("ACL.GetPolicies", &req, &resp); err != nil {
+		// If we encounter an error but have cached policies, mask the error and extend the cache
+		if len(missing) == 0 {
+			c.logger.Printf("[WARN] client: failed to resolve policies, using expired cached value: %v", err)
+			out = append(out, expired...)
+			return out, nil
+		}
+		return nil, err
+	}
+
+	// Handle each output
+	for _, policy := range resp.Policies {
+		c.policyCache.Add(policy.Name, &cachedACLValue{
+			Policy:    policy,
+			CacheTime: time.Now(),
+		})
+		out = append(out, policy)
+	}
+
+	// Return the valid policies
+	return out, nil
+}

client/acl_test.go

@@ -0,0 +1,164 @@
+package client
+
+import (
+	"testing"
+
+	"github.com/hashicorp/nomad/acl"
+	"github.com/hashicorp/nomad/client/config"
+	"github.com/hashicorp/nomad/nomad/mock"
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/hashicorp/nomad/testutil"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestClient_ACL_resolveTokenValue(t *testing.T) {
+	s1, _ := testServer(t, nil)
+	defer s1.Shutdown()
+	testutil.WaitForLeader(t, s1.RPC)
+
+	c1 := testClient(t, func(c *config.Config) {
+		c.RPCHandler = s1
+	})
+	defer c1.Shutdown()
+
+	// Create a policy / token
+	policy := mock.ACLPolicy()
+	policy2 := mock.ACLPolicy()
+	token := mock.ACLToken()
+	token.Policies = []string{policy.Name, policy2.Name}
+	token2 := mock.ACLToken()
+	token2.Type = structs.ACLManagementToken
+	token2.Policies = nil
+	err := s1.State().UpsertACLPolicies(100, []*structs.ACLPolicy{policy, policy2})
+	assert.Nil(t, err)
+	err = s1.State().UpsertACLTokens(110, []*structs.ACLToken{token, token2})
+	assert.Nil(t, err)
+
+	// Test the client resolution
+	out0, err := c1.resolveTokenValue("")
+	assert.Nil(t, err)
+	assert.NotNil(t, out0)
+	assert.Equal(t, structs.AnonymousACLToken, out0)
+
+	// Test the client resolution
+	out1, err := c1.resolveTokenValue(token.SecretID)
+	assert.Nil(t, err)
+	assert.NotNil(t, out1)
+	assert.Equal(t, token, out1)
+
+	out2, err := c1.resolveTokenValue(token2.SecretID)
+	assert.Nil(t, err)
+	assert.NotNil(t, out2)
+	assert.Equal(t, token2, out2)
+
+	out3, err := c1.resolveTokenValue(token.SecretID)
+	assert.Nil(t, err)
+	assert.NotNil(t, out3)
+	if out1 != out3 {
+		t.Fatalf("bad caching")
+	}
+}
+
+func TestClient_ACL_resolvePolicies(t *testing.T) {
+	s1, _ := testServer(t, nil)
+	defer s1.Shutdown()
+	testutil.WaitForLeader(t, s1.RPC)
+
+	c1 := testClient(t, func(c *config.Config) {
+		c.RPCHandler = s1
+	})
+	defer c1.Shutdown()
+
+	// Create a policy / token
+	policy := mock.ACLPolicy()
+	policy2 := mock.ACLPolicy()
+	token := mock.ACLToken()
+	token.Policies = []string{policy.Name, policy2.Name}
+	token2 := mock.ACLToken()
+	token2.Type = structs.ACLManagementToken
+	token2.Policies = nil
+	err := s1.State().UpsertACLPolicies(100, []*structs.ACLPolicy{policy, policy2})
+	assert.Nil(t, err)
+	err = s1.State().UpsertACLTokens(110, []*structs.ACLToken{token, token2})
+	assert.Nil(t, err)
+
+	// Test the client resolution
+	out, err := c1.resolvePolicies([]string{policy.Name, policy2.Name})
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(out))
+
+	// Test caching
+	out2, err := c1.resolvePolicies([]string{policy.Name, policy2.Name})
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(out2))
+
+	// Check we get the same objects back (ignore ordering)
+	if out[0] != out2[0] && out[0] != out2[1] {
+		t.Fatalf("bad caching")
+	}
+}
+
+func TestClient_ACL_resolveToken_Disabled(t *testing.T) {
+	s1, _ := testServer(t, nil)
+	defer s1.Shutdown()
+	testutil.WaitForLeader(t, s1.RPC)
+
+	c1 := testClient(t, func(c *config.Config) {
+		c.RPCHandler = s1
+	})
+	defer c1.Shutdown()
+
+	// Should always get nil when disabled
+	aclObj, err := c1.resolveToken("blah")
+	assert.Nil(t, err)
+	assert.Nil(t, aclObj)
+}
+
+func TestClient_ACL_resolveToken(t *testing.T) {
+	s1, _ := testServer(t, nil)
+	defer s1.Shutdown()
+	testutil.WaitForLeader(t, s1.RPC)
+
+	c1 := testClient(t, func(c *config.Config) {
+		c.RPCHandler = s1
+		c.ACLEnabled = true
+	})
+	defer c1.Shutdown()
+
+	// Create a policy / token
+	policy := mock.ACLPolicy()
+	policy2 := mock.ACLPolicy()
+	token := mock.ACLToken()
+	token.Policies = []string{policy.Name, policy2.Name}
+	token2 := mock.ACLToken()
+	token2.Type = structs.ACLManagementToken
+	token2.Policies = nil
+	err := s1.State().UpsertACLPolicies(100, []*structs.ACLPolicy{policy, policy2})
+	assert.Nil(t, err)
+	err = s1.State().UpsertACLTokens(110, []*structs.ACLToken{token, token2})
+	assert.Nil(t, err)
+
+	// Test the client resolution
+	out, err := c1.resolveToken(token.SecretID)
+	assert.Nil(t, err)
+	assert.NotNil(t, out)
+
+	// Test caching
+	out2, err := c1.resolveToken(token.SecretID)
+	assert.Nil(t, err)
+	if out != out2 {
+		t.Fatalf("should be cached")
+	}
+
+	// Test management token
+	out3, err := c1.resolveToken(token2.SecretID)
+	assert.Nil(t, err)
+	if acl.ManagementACL != out3 {
+		t.Fatalf("should be management")
+	}
+
+	// Test bad token
+	out4, err := c1.resolveToken(structs.GenerateUUID())
+	assert.Equal(t, structs.TokenNotFound, err)
+	assert.Nil(t, out4)
+}