Fix legacy management tokens in unupgraded secondary dcs (#7908)

The ACL.GetPolicy RPC endpoint was supposed to return the “parent” policy and not always the default policy. In the case of legacy management tokens the parent policy was supposed to be “manage”. The result of us not sending this properly was that operations that required specifically a management token such as saving a snapshot would not work in secondary DCs until they were upgraded.

agent/consul/acl.go

@@ -1154,16 +1154,17 @@ func (r *ACLResolver) ACLsEnabled() bool {
 	return true
 }
 
-func (r *ACLResolver) GetMergedPolicyForToken(token string) (*acl.Policy, error) {
-	policies, err := r.resolveTokenToPolicies(token)
+func (r *ACLResolver) GetMergedPolicyForToken(token string) (structs.ACLIdentity, *acl.Policy, error) {
+	ident, policies, err := r.resolveTokenToIdentityAndPolicies(token)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	if len(policies) == 0 {
-		return nil, acl.ErrNotFound
+		return nil, nil, acl.ErrNotFound
 	}
 
-	return policies.Merge(r.cache, r.aclConf)
+	policy, err := policies.Merge(r.cache, r.aclConf)
+	return ident, policy, err
 }
 
 // aclFilter is used to filter results from our state store based on ACL rules

agent/consul/acl_endpoint.go

@@ -1321,11 +1321,15 @@ func (a *ACL) GetPolicy(args *structs.ACLPolicyResolveLegacyRequest, reply *stru
 	// Get the policy via the cache
 	parent := a.srv.config.ACLDefaultPolicy
 
-	policy, err := a.srv.acls.GetMergedPolicyForToken(args.ACL)
+	ident, policy, err := a.srv.acls.GetMergedPolicyForToken(args.ACL)
 	if err != nil {
 		return err
 	}
 
+	if token, ok := ident.(*structs.ACLToken); ok && token.Type == structs.ACLTokenTypeManagement {
+		parent = "manage"
+	}
+
 	// translates the structures internals to most closely match what could be expressed in the original rule language
 	policy = policy.ConvertToLegacy()
 

agent/consul/acl_endpoint_test.go

@@ -453,6 +453,7 @@ func TestACLEndpoint_GetPolicy(t *testing.T) {
 		c.ACLDatacenter = "dc1"
 		c.ACLsEnabled = true
 		c.ACLMasterToken = "root"
+		c.ACLDefaultPolicy = "deny"
 	})
 	defer os.RemoveAll(dir1)
 	defer s1.Shutdown()
@@ -471,9 +472,7 @@ func TestACLEndpoint_GetPolicy(t *testing.T) {
 		WriteRequest: structs.WriteRequest{Token: "root"},
 	}
 	var out string
-	if err := msgpackrpc.CallWithCodec(codec, "ACL.Apply", &arg, &out); err != nil {
-		t.Fatalf("err: %v", err)
-	}
+	require.NoError(t, msgpackrpc.CallWithCodec(codec, "ACL.Apply", &arg, &out))
 
 	getR := structs.ACLPolicyResolveLegacyRequest{
 		Datacenter: "dc1",
@@ -482,32 +481,43 @@ func TestACLEndpoint_GetPolicy(t *testing.T) {
 
 	var acls structs.ACLPolicyResolveLegacyResponse
 	retry.Run(t, func(r *retry.R) {
-
-		if err := msgpackrpc.CallWithCodec(codec, "ACL.GetPolicy", &getR, &acls); err != nil {
-			r.Fatalf("err: %v", err)
-		}
-
-		if acls.Policy == nil {
-			r.Fatalf("Bad: %v", acls)
-		}
-		if acls.TTL != 30*time.Second {
-			r.Fatalf("bad: %v", acls)
-		}
+		require.NoError(r, msgpackrpc.CallWithCodec(codec, "ACL.GetPolicy", &getR, &acls))
+		require.NotNil(t, acls.Policy)
+		require.Equal(t, "deny", acls.Parent)
+		require.Equal(t, 30*time.Second, acls.TTL)
 	})
 
 	// Do a conditional lookup with etag
 	getR.ETag = acls.ETag
 	var out2 structs.ACLPolicyResolveLegacyResponse
-	if err := msgpackrpc.CallWithCodec(codec, "ACL.GetPolicy", &getR, &out2); err != nil {
-		t.Fatalf("err: %v", err)
+	require.NoError(t, msgpackrpc.CallWithCodec(codec, "ACL.GetPolicy", &getR, &out2))
+
+	require.Nil(t, out2.Policy)
+	require.Equal(t, 30*time.Second, out2.TTL)
+}
+
+func TestACLEndpoint_GetPolicy_Management(t *testing.T) {
+	t.Parallel()
+	dir1, s1 := testServerWithConfig(t, testServerACLConfig(nil))
+	defer os.RemoveAll(dir1)
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	defer codec.Close()
+
+	// wait for leader election and leader establishment to finish.
+	// after this the global management policy, master token and
+	// anonymous token will have been injected into the state store
+	// and we will be ready to resolve the master token
+	waitForLeaderEstablishment(t, s1)
+
+	req := structs.ACLPolicyResolveLegacyRequest{
+		Datacenter: s1.config.Datacenter,
+		ACL:        TestDefaultMasterToken,
 	}
 
-	if out2.Policy != nil {
-		t.Fatalf("Bad: %v", out2)
-	}
-	if out2.TTL != 30*time.Second {
-		t.Fatalf("bad: %v", out2)
-	}
+	var resp structs.ACLPolicyResolveLegacyResponse
+	require.NoError(t, msgpackrpc.CallWithCodec(codec, "ACL.GetPolicy", &req, &resp))
+	require.Equal(t, "manage", resp.Parent)
 }
 
 func TestACLEndpoint_List(t *testing.T) {