add missing code and fix enterprise specific code (#15375)

* add missing code and fix enterprise specific code

* fix retry

* fix flaky tests

* fix linter error in test

acl/MockAuthorizer.go

@@ -6,6 +6,16 @@ type MockAuthorizer struct {
 	mock.Mock
 }
 
+func (m *MockAuthorizer) NamespaceRead(s string, ctx *AuthorizerContext) EnforcementDecision {
+	ret := m.Called(s, ctx)
+	return ret.Get(0).(EnforcementDecision)
+}
+
+func (m *MockAuthorizer) NamespaceWrite(s string, ctx *AuthorizerContext) EnforcementDecision {
+	ret := m.Called(s, ctx)
+	return ret.Get(0).(EnforcementDecision)
+}
+
 var _ Authorizer = (*MockAuthorizer)(nil)
 
 // ACLRead checks for permission to list all the ACLs

agent/consul/operator_backend.go

@@ -21,7 +21,14 @@ func NewOperatorBackend(srv *Server) *OperatorBackend {
 }
 
 func (op *OperatorBackend) ResolveTokenAndDefaultMeta(token string, entMeta *acl.EnterpriseMeta, authzCtx *acl.AuthorizerContext) (resolver.Result, error) {
-	return op.srv.ResolveTokenAndDefaultMeta(token, entMeta, authzCtx)
+	res, err := op.srv.ResolveTokenAndDefaultMeta(token, entMeta, authzCtx)
+	if err != nil {
+		return resolver.Result{}, err
+	}
+	if err := op.srv.validateEnterpriseToken(res.ACLIdentity); err != nil {
+		return resolver.Result{}, err
+	}
+	return res, err
 }
 
 func (op *OperatorBackend) TransferLeader(_ context.Context, request *pboperator.TransferLeaderRequest) (*pboperator.TransferLeaderResponse, error) {

agent/consul/operator_backend_test.go

@@ -60,17 +60,14 @@ func TestOperatorBackend_TransferLeader(t *testing.T) {
 		reply, err := operatorClient.TransferLeader(ctx, &req)
 		require.NoError(t, err)
 		require.True(t, reply.Success)
-		time.Sleep(1 * time.Second)
 		testrpc.WaitForLeader(t, s1.RPC, "dc1")
 		retry.Run(t, func(r *retry.R) {
+			time.Sleep(1 * time.Second)
 			afterLeader, _ := s1.raft.LeaderWithID()
 			require.NotEmpty(r, afterLeader)
+			require.NotEqual(r, afterLeader, beforeLeader)
 		})
-		afterLeader, _ := s1.raft.LeaderWithID()
-		require.NotEmpty(t, afterLeader)
-		if afterLeader == beforeLeader {
-			t.Fatalf("leader should have changed %s == %s", afterLeader, beforeLeader)
-		}
+
 	})
 }
 
@@ -94,7 +91,6 @@ func TestOperatorBackend_TransferLeaderWithACL(t *testing.T) {
 	s1 := nodes.Servers[0]
 	// Make sure a leader is elected
 	testrpc.WaitForLeader(t, s1.RPC, "dc1")
-
 	// Make a write call to server2 and make sure it gets forwarded to server1
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	t.Cleanup(cancel)
@@ -109,6 +105,13 @@ func TestOperatorBackend_TransferLeaderWithACL(t *testing.T) {
 
 	operatorClient := pboperator.NewOperatorServiceClient(conn)
 
+	codec := rpcClient(t, s1)
+	rules := `operator = "write"`
+	tokenWrite := createTokenWithPolicyNameFull(t, codec, "the-policy-write", rules, "root")
+	rules = `operator = "read"`
+	tokenRead := createToken(t, codec, rules)
+	require.NoError(t, err)
+
 	testutil.RunStep(t, "transfer leader no token", func(t *testing.T) {
 		beforeLeader, _ := s1.raft.LeaderWithID()
 		require.NotEmpty(t, beforeLeader)
@@ -122,14 +125,14 @@ func TestOperatorBackend_TransferLeaderWithACL(t *testing.T) {
 		time.Sleep(1 * time.Second)
 		testrpc.WaitForLeader(t, s1.RPC, "dc1")
 		retry.Run(t, func(r *retry.R) {
+			time.Sleep(1 * time.Second)
 			afterLeader, _ := s1.raft.LeaderWithID()
 			require.NotEmpty(r, afterLeader)
+			if afterLeader != beforeLeader {
+				r.Fatalf("leader should have changed %s == %s", afterLeader, beforeLeader)
+			}
 		})
-		afterLeader, _ := s1.raft.LeaderWithID()
-		require.NotEmpty(t, afterLeader)
-		if afterLeader != beforeLeader {
-			t.Fatalf("leader should have changed %s == %s", afterLeader, beforeLeader)
-		}
+
 	})
 
 	testutil.RunStep(t, "transfer leader operator read token", func(t *testing.T) {
@@ -140,26 +143,22 @@ func TestOperatorBackend_TransferLeaderWithACL(t *testing.T) {
 		req := pboperator.TransferLeaderRequest{
 			ID: "",
 		}
-		codec := rpcClient(t, s1)
-		rules := `operator = "read"`
-		tokenRead := createToken(t, codec, rules)
 
 		ctxToken, err := external.ContextWithQueryOptions(ctx, structs.QueryOptions{Token: tokenRead})
 		require.NoError(t, err)
+
 		reply, err := operatorClient.TransferLeader(ctxToken, &req)
 		require.True(t, acl.IsErrPermissionDenied(err))
 		require.Nil(t, reply)
-		time.Sleep(1 * time.Second)
 		testrpc.WaitForLeader(t, s1.RPC, "dc1")
 		retry.Run(t, func(r *retry.R) {
+			time.Sleep(1 * time.Second)
 			afterLeader, _ := s1.raft.LeaderWithID()
 			require.NotEmpty(r, afterLeader)
+			if afterLeader != beforeLeader {
+				r.Fatalf("leader should have changed %s == %s", afterLeader, beforeLeader)
+			}
 		})
-		afterLeader, _ := s1.raft.LeaderWithID()
-		require.NotEmpty(t, afterLeader)
-		if afterLeader != beforeLeader {
-			t.Fatalf("leader should have changed %s == %s", afterLeader, beforeLeader)
-		}
 	})
 
 	testutil.RunStep(t, "transfer leader operator write token", func(t *testing.T) {
@@ -170,9 +169,6 @@ func TestOperatorBackend_TransferLeaderWithACL(t *testing.T) {
 		req := pboperator.TransferLeaderRequest{
 			ID: "",
 		}
-		codec := rpcClient(t, s1)
-		rules := `operator = "write"`
-		tokenWrite := createTokenWithPolicyNameFull(t, codec, "the-policy-write", rules, "root")
 		ctxToken, err := external.ContextWithQueryOptions(ctx, structs.QueryOptions{Token: tokenWrite.SecretID})
 		require.NoError(t, err)
 		reply, err := operatorClient.TransferLeader(ctxToken, &req)
@@ -181,13 +177,12 @@ func TestOperatorBackend_TransferLeaderWithACL(t *testing.T) {
 		time.Sleep(1 * time.Second)
 		testrpc.WaitForLeader(t, s1.RPC, "dc1")
 		retry.Run(t, func(r *retry.R) {
+			time.Sleep(1 * time.Second)
 			afterLeader, _ := s1.raft.LeaderWithID()
 			require.NotEmpty(r, afterLeader)
+			if afterLeader == beforeLeader {
+				r.Fatalf("leader should have changed %s == %s", afterLeader, beforeLeader)
+			}
 		})
-		afterLeader, _ := s1.raft.LeaderWithID()
-		require.NotEmpty(t, afterLeader)
-		if afterLeader == beforeLeader {
-			t.Fatalf("leader should have changed %s == %s", afterLeader, beforeLeader)
-		}
 	})
 }