acl: global tokens created by auth methods now correctly replicate to secondary datacenters

agent/consul/acl_endpoint_test.go

@@ -4775,19 +4775,32 @@ func TestACLEndpoint_Login_with_MaxTokenTTL(t *testing.T) {
 func TestACLEndpoint_Login_with_TokenLocality(t *testing.T) {
 	go t.Parallel()
 
-	dir1, s1 := testServerWithConfig(t, func(c *Config) {
-		c.ACLDatacenter = "dc1"
-		c.ACLsEnabled = true
-		c.ACLMasterToken = "root"
-	})
-	defer os.RemoveAll(dir1)
-	defer s1.Shutdown()
-	codec := rpcClient(t, s1)
-	defer codec.Close()
+	_, s1, codec := testACLServerWithConfig(t, func(c *Config) {
+		c.ACLTokenMinExpirationTTL = 10 * time.Millisecond
+		c.ACLTokenMaxExpirationTTL = 5 * time.Second
+	}, false)
+
+	_, s2, _ := testACLServerWithConfig(t, func(c *Config) {
+		c.Datacenter = "dc2"
+		c.ACLTokenMinExpirationTTL = 10 * time.Millisecond
+		c.ACLTokenMaxExpirationTTL = 5 * time.Second
+		// token replication is required to test deleting non-local tokens in secondary dc
+		c.ACLTokenReplication = true
+	}, true)
 
 	waitForLeaderEstablishment(t, s1)
+	waitForLeaderEstablishment(t, s2)
+
+	joinWAN(t, s2, s1)
+
+	waitForNewACLs(t, s1)
+	waitForNewACLs(t, s2)
+
+	// Ensure s2 is authoritative.
+	waitForNewACLReplication(t, s2, structs.ACLReplicateTokens, 1, 1, 0)
 
 	acl := ACL{srv: s1}
+	acl2 := ACL{srv: s2}
 
 	testSessionID := testauth.StartSession()
 	defer testauth.ResetSession(testSessionID)
@@ -4799,18 +4812,20 @@ func TestACLEndpoint_Login_with_TokenLocality(t *testing.T) {
 	)
 
 	cases := map[string]struct {
-		tokenLocality string
-		expectLocal   bool
+		tokenLocality     string
+		expectLocal       bool
+		deleteFromReplica bool
 	}{
-		"empty":  {tokenLocality: "", expectLocal: true},
-		"local":  {tokenLocality: "local", expectLocal: true},
-		"global": {tokenLocality: "global", expectLocal: false},
+		"empty":             {tokenLocality: "", expectLocal: true},
+		"local":             {tokenLocality: "local", expectLocal: true},
+		"global dc1 delete": {tokenLocality: "global", expectLocal: false, deleteFromReplica: false},
+		"global dc2 delete": {tokenLocality: "global", expectLocal: false, deleteFromReplica: true},
 	}
 
 	for name, tc := range cases {
 		tc := tc
 		t.Run(name, func(t *testing.T) {
-			method, err := upsertTestCustomizedAuthMethod(codec, "root", "dc1", func(method *structs.ACLAuthMethod) {
+			method, err := upsertTestCustomizedAuthMethod(codec, TestDefaultMasterToken, "dc1", func(method *structs.ACLAuthMethod) {
 				method.TokenLocality = tc.tokenLocality
 				method.Config = map[string]interface{}{
 					"SessionID": testSessionID,
@@ -4819,7 +4834,7 @@ func TestACLEndpoint_Login_with_TokenLocality(t *testing.T) {
 			require.NoError(t, err)
 
 			_, err = upsertTestBindingRule(
-				codec, "root", "dc1", method.Name,
+				codec, TestDefaultMasterToken, "dc1", method.Name,
 				"",
 				structs.BindingRuleBindTypeService,
 				"web",
@@ -4841,7 +4856,7 @@ func TestACLEndpoint_Login_with_TokenLocality(t *testing.T) {
 
 			secretID := resp.SecretID
 
-			got := &resp
+			got := resp.Clone()
 			got.CreateIndex = 0
 			got.ModifyIndex = 0
 			got.AccessorID = ""
@@ -4863,13 +4878,38 @@ func TestACLEndpoint_Login_with_TokenLocality(t *testing.T) {
 			require.Equal(t, got, expect)
 
 			// Now turn around and nuke it.
+			var (
+				logoutACL ACL
+				logoutDC  string
+			)
+			if tc.deleteFromReplica {
+				// wait for it to show up
+				retry.Run(t, func(r *retry.R) {
+					var resp structs.ACLTokenResponse
+					require.NoError(r, acl2.TokenRead(&structs.ACLTokenGetRequest{
+						TokenID:        secretID,
+						TokenIDType:    structs.ACLTokenSecret,
+						Datacenter:     "dc2",
+						EnterpriseMeta: *defaultEntMeta,
+						QueryOptions:   structs.QueryOptions{Token: TestDefaultMasterToken},
+					}, &resp))
+					require.NotNil(r, resp.Token, "cannot lookup token with secretID %q", secretID)
+				})
+
+				logoutACL = acl2
+				logoutDC = "dc2"
+			} else {
+				logoutACL = acl
+				logoutDC = "dc1"
+			}
+
 			logoutReq := structs.ACLLogoutRequest{
-				Datacenter:   "dc1",
+				Datacenter:   logoutDC,
 				WriteRequest: structs.WriteRequest{Token: secretID},
 			}
 
 			var ignored bool
-			require.NoError(t, acl.Logout(&logoutReq, &ignored))
+			require.NoError(t, logoutACL.Logout(&logoutReq, &ignored))
 		})
 	}
 }

agent/consul/acl_replication_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/agent/consul/authmethod/testauth"
 	"github.com/hashicorp/consul/agent/structs"
 	tokenStore "github.com/hashicorp/consul/agent/token"
 	"github.com/hashicorp/consul/sdk/testutil/retry"
@@ -349,6 +350,32 @@ func TestACLReplication_Tokens(t *testing.T) {
 		tokens = append(tokens, &token)
 	}
 
+	// Create an auth method in the primary that can create global tokens
+	// so that we ensure that these replicate correctly.
+	testSessionID := testauth.StartSession()
+	defer testauth.ResetSession(testSessionID)
+	testauth.InstallSessionToken(testSessionID, "fake-token", "default", "demo", "abc123")
+	method1, err := upsertTestCustomizedAuthMethod(client, "root", "dc1", func(method *structs.ACLAuthMethod) {
+		method.TokenLocality = "global"
+		method.Config = map[string]interface{}{
+			"SessionID": testSessionID,
+		}
+	})
+	require.NoError(t, err)
+	_, err = upsertTestBindingRule(client, "root", "dc1", method1.Name, "", structs.BindingRuleBindTypeService, "demo")
+	require.NoError(t, err)
+
+	// Create one token via this process.
+	methodToken := structs.ACLToken{}
+	require.NoError(t, s1.RPC("ACL.Login", &structs.ACLLoginRequest{
+		Auth: &structs.ACLLoginParams{
+			AuthMethod:  method1.Name,
+			BearerToken: "fake-token",
+		},
+		Datacenter: "dc1",
+	}, &methodToken))
+	tokens = append(tokens, &methodToken)
+
 	checkSame := func(t *retry.R) {
 		// only account for global tokens - local tokens shouldn't be replicated
 		index, remote, err := s1.fsm.State().ACLTokenList(nil, false, true, "", "", "", nil, nil)
@@ -359,6 +386,11 @@ func TestACLReplication_Tokens(t *testing.T) {
 		require.Len(t, local, len(remote))
 		for i, token := range remote {
 			require.Equal(t, token.Hash, local[i].Hash)
+
+			if token.AccessorID == methodToken.AccessorID {
+				require.Equal(t, method1.Name, token.AuthMethod)
+				require.Equal(t, method1.Name, local[i].AuthMethod)
+			}
 		}
 
 		s2.aclReplicationStatusLock.RLock()

agent/consul/acl_replication_types.go

@@ -113,6 +113,7 @@ func (r *aclTokenReplicator) UpdateLocalBatch(ctx context.Context, srv *Server,
 		Tokens:            r.updated[start:end],
 		CAS:               false,
 		AllowMissingLinks: true,
+		FromReplication:   true,
 	}
 
 	resp, err := srv.raftApply(structs.ACLTokenSetRequestType, &req)

agent/consul/fsm/commands_oss.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/armon/go-metrics"
 	"github.com/armon/go-metrics/prometheus"
+	"github.com/hashicorp/consul/agent/consul/state"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/api"
 )
@@ -509,7 +510,14 @@ func (c *FSM) applyACLTokenSetOperation(buf []byte, index uint64) interface{} {
 	defer metrics.MeasureSinceWithLabels([]string{"fsm", "acl", "token"}, time.Now(),
 		[]metrics.Label{{Name: "op", Value: "upsert"}})
 
-	return c.state.ACLTokenBatchSet(index, req.Tokens, req.CAS, req.AllowMissingLinks, req.ProhibitUnprivileged)
+	opts := state.ACLTokenSetOptions{
+		CAS:                          req.CAS,
+		AllowMissingPolicyAndRoleIDs: req.AllowMissingLinks,
+		ProhibitUnprivileged:         req.ProhibitUnprivileged,
+		Legacy:                       false,
+		FromReplication:              req.FromReplication,
+	}
+	return c.state.ACLTokenBatchSet(index, req.Tokens, opts)
 }
 
 func (c *FSM) applyACLTokenDeleteOperation(buf []byte, index uint64) interface{} {

agent/consul/state/acl.go

@@ -305,7 +305,7 @@ func (s *Store) ACLBootstrap(idx, resetIndex uint64, token *structs.ACLToken, le
 		}
 	}
 
-	if err := aclTokenSetTxn(tx, idx, token, false, false, false, legacy); err != nil {
+	if err := aclTokenSetTxn(tx, idx, token, ACLTokenSetOptions{Legacy: legacy}); err != nil {
 		return fmt.Errorf("failed inserting bootstrap token: %v", err)
 	}
 	if err := tx.Insert("index", &IndexEntry{"acl-token-bootstrap", idx}); err != nil {
@@ -632,19 +632,31 @@ func (s *Store) ACLTokenSet(idx uint64, token *structs.ACLToken, legacy bool) er
 	defer tx.Abort()
 
 	// Call set on the ACL
-	if err := aclTokenSetTxn(tx, idx, token, false, false, false, legacy); err != nil {
+	if err := aclTokenSetTxn(tx, idx, token, ACLTokenSetOptions{Legacy: legacy}); err != nil {
 		return err
 	}
 
 	return tx.Commit()
 }
 
-func (s *Store) ACLTokenBatchSet(idx uint64, tokens structs.ACLTokens, cas, allowMissingPolicyAndRoleIDs, prohibitUnprivileged bool) error {
+type ACLTokenSetOptions struct {
+	CAS                          bool
+	AllowMissingPolicyAndRoleIDs bool
+	ProhibitUnprivileged         bool
+	Legacy                       bool
+	FromReplication              bool
+}
+
+func (s *Store) ACLTokenBatchSet(idx uint64, tokens structs.ACLTokens, opts ACLTokenSetOptions) error {
+	if opts.Legacy {
+		return fmt.Errorf("failed inserting acl token: cannot use this endpoint to persist legacy tokens")
+	}
+
 	tx := s.db.WriteTxn(idx)
 	defer tx.Abort()
 
 	for _, token := range tokens {
-		if err := aclTokenSetTxn(tx, idx, token, cas, allowMissingPolicyAndRoleIDs, prohibitUnprivileged, false); err != nil {
+		if err := aclTokenSetTxn(tx, idx, token, opts); err != nil {
 			return err
 		}
 	}
@@ -654,16 +666,20 @@ func (s *Store) ACLTokenBatchSet(idx uint64, tokens structs.ACLTokens, cas, allo
 
 // aclTokenSetTxn is the inner method used to insert an ACL token with the
 // proper indexes into the state store.
-func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas, allowMissingPolicyAndRoleIDs, prohibitUnprivileged, legacy bool) error {
+func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, opts ACLTokenSetOptions) error {
 	// Check that the ID is set
 	if token.SecretID == "" {
 		return ErrMissingACLTokenSecret
 	}
 
-	if !legacy && token.AccessorID == "" {
+	if !opts.Legacy && token.AccessorID == "" {
 		return ErrMissingACLTokenAccessor
 	}
 
+	if opts.FromReplication && token.Local {
+		return fmt.Errorf("Cannot replicate local tokens")
+	}
+
 	// DEPRECATED (ACL-Legacy-Compat)
 	if token.Rules != "" {
 		// When we update a legacy acl token we may have to correct old HCL to
@@ -688,7 +704,7 @@ func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas, allowMiss
 		original = existing.(*structs.ACLToken)
 	}
 
-	if cas {
+	if opts.CAS {
 		// set-if-unset case
 		if token.ModifyIndex == 0 && original != nil {
 			return nil
@@ -703,7 +719,7 @@ func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas, allowMiss
 		}
 	}
 
-	if legacy && original != nil {
+	if opts.Legacy && original != nil {
 		if original.UsesNonLegacyFields() {
 			return fmt.Errorf("failed inserting acl token: cannot use legacy endpoint to modify a non-legacy token")
 		}
@@ -716,16 +732,16 @@ func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas, allowMiss
 	}
 
 	var numValidPolicies int
-	if numValidPolicies, err = resolveTokenPolicyLinks(tx, token, allowMissingPolicyAndRoleIDs); err != nil {
+	if numValidPolicies, err = resolveTokenPolicyLinks(tx, token, opts.AllowMissingPolicyAndRoleIDs); err != nil {
 		return err
 	}
 
 	var numValidRoles int
-	if numValidRoles, err = resolveTokenRoleLinks(tx, token, allowMissingPolicyAndRoleIDs); err != nil {
+	if numValidRoles, err = resolveTokenRoleLinks(tx, token, opts.AllowMissingPolicyAndRoleIDs); err != nil {
 		return err
 	}
 
-	if token.AuthMethod != "" {
+	if token.AuthMethod != "" && !opts.FromReplication {
 		method, err := getAuthMethodWithTxn(tx, nil, token.AuthMethod, token.ACLAuthMethodEnterpriseMeta.ToEnterpriseMeta())
 		if err != nil {
 			return err
@@ -749,7 +765,7 @@ func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas, allowMiss
 		}
 	}
 
-	if prohibitUnprivileged {
+	if opts.ProhibitUnprivileged {
 		if numValidRoles == 0 && numValidPolicies == 0 && len(token.ServiceIdentities) == 0 && len(token.NodeIdentities) == 0 {
 			return ErrTokenHasNoPrivileges
 		}
@@ -1059,7 +1075,7 @@ func aclTokenDeleteTxn(tx *txn, idx uint64, value, index string, entMeta *struct
 	return aclTokenDeleteWithToken(tx, token.(*structs.ACLToken), idx)
 }
 
-func aclTokenDeleteAllForAuthMethodTxn(tx *txn, idx uint64, methodName string, methodMeta *structs.EnterpriseMeta) error {
+func aclTokenDeleteAllForAuthMethodTxn(tx *txn, idx uint64, methodName string, methodGlobalLocality bool, methodMeta *structs.EnterpriseMeta) error {
 	// collect all the tokens linked with the given auth method.
 	iter, err := aclTokenListByAuthMethod(tx, methodName, methodMeta, structs.WildcardEnterpriseMeta())
 	if err != nil {
@@ -1069,7 +1085,15 @@ func aclTokenDeleteAllForAuthMethodTxn(tx *txn, idx uint64, methodName string, m
 	var tokens structs.ACLTokens
 	for raw := iter.Next(); raw != nil; raw = iter.Next() {
 		token := raw.(*structs.ACLToken)
-		tokens = append(tokens, token)
+		tokenIsGlobal := !token.Local
+
+		// Need to ensure that if we have an auth method named "blah" in the
+		// primary and secondary datacenters, and the primary instance has
+		// TokenLocality==global that when we delete the secondary instance we
+		// don't also blow away replicated tokens from the primary.
+		if methodGlobalLocality == tokenIsGlobal {
+			tokens = append(tokens, token)
+		}
 	}
 
 	if len(tokens) > 0 {
@@ -1877,7 +1901,7 @@ func aclAuthMethodDeleteTxn(tx *txn, idx uint64, name string, entMeta *structs.E
 		return err
 	}
 
-	if err := aclTokenDeleteAllForAuthMethodTxn(tx, idx, method.Name, entMeta); err != nil {
+	if err := aclTokenDeleteAllForAuthMethodTxn(tx, idx, method.Name, method.TokenLocality == "global", entMeta); err != nil {
 		return err
 	}