acl: fix a bug where roles could be duplicated by name.

An ACL roles name must be unique, however, a bug meant multiple
roles of the same same could be created. This fixes that problem
with checks in the RPC handler and state store.

api/acl_test.go

@@ -330,7 +330,7 @@ func TestACLTokens_Info(t *testing.T) {
 				// Create an ACL role referencing the previously created
 				// policy.
 				role := ACLRole{
-					Name:     "acl-role-api-test",
+					Name:     "acl-role-api-test-role-and-policy",
 					Policies: []*ACLRolePolicyLink{{Name: aclPolicy1.Name}},
 				}
 				aclRoleCreateResp, writeMeta, err := testClient.ACLRoles().Create(&role, nil)
@@ -341,7 +341,7 @@ func TestACLTokens_Info(t *testing.T) {
 
 				// Create a token with a role linking.
 				token := &ACLToken{
-					Name:     "token-with-role-link",
+					Name:     "token-with-role-and-policy-link",
 					Type:     "client",
 					Policies: []string{aclPolicy2.Name},
 					Roles:    []*ACLTokenRoleLink{{Name: role.Name}},

nomad/acl_endpoint.go

@@ -1122,6 +1122,28 @@ func (a *ACL) UpsertRoles(
 			return structs.NewErrRPCCodedf(http.StatusBadRequest, "role %d invalid: %v", idx, err)
 		}
 
+		// If the caller has passed a role ID, this call is considered an
+		// update to an existing role. We should therefore ensure it is found
+		// within state. Otherwise, the call is considered a new creation, and
+		// we must ensure a role of the same name does not exist.
+		if role.ID == "" {
+			existingRole, err := stateSnapshot.GetACLRoleByName(nil, role.Name)
+			if err != nil {
+				return structs.NewErrRPCCodedf(http.StatusInternalServerError, "role lookup failed: %v", err)
+			}
+			if existingRole != nil {
+				return structs.NewErrRPCCodedf(http.StatusBadRequest, "role with name %s already exists", role.Name)
+			}
+		} else {
+			existingRole, err := stateSnapshot.GetACLRoleByID(nil, role.ID)
+			if err != nil {
+				return structs.NewErrRPCCodedf(http.StatusInternalServerError, "role lookup failed: %v", err)
+			}
+			if existingRole == nil {
+				return structs.NewErrRPCCodedf(http.StatusBadRequest, "cannot find role %s", role.ID)
+			}
+		}
+
 		policyNames := make(map[string]struct{})
 		var policiesLinks []*structs.ACLRolePolicyLink
 
@@ -1156,19 +1178,6 @@ func (a *ACL) UpsertRoles(
 		// Stored the potentially updated policy links within our role.
 		role.Policies = policiesLinks
 
-		// If the caller has passed a role ID, this call is considered an
-		// update to an existing role. We should therefore ensure it is found
-		// within state.
-		if role.ID != "" {
-			out, err := stateSnapshot.GetACLRoleByID(nil, role.ID)
-			if err != nil {
-				return structs.NewErrRPCCodedf(http.StatusBadRequest, "role lookup failed: %v", err)
-			}
-			if out == nil {
-				return structs.NewErrRPCCodedf(http.StatusBadRequest, "cannot find role %s", role.ID)
-			}
-		}
-
 		role.Canonicalize()
 		role.SetHash()
 	}

nomad/acl_endpoint_test.go

@@ -1998,6 +1998,22 @@ func TestACL_UpsertRoles(t *testing.T) {
 	err = msgpackrpc.CallWithCodec(codec, structs.ACLUpsertRolesRPCMethod, aclRoleReq5, &aclRoleResp5)
 	require.Error(t, err)
 	require.NotContains(t, err, "Permission denied")
+
+	// Try and create a role with a name that already exists within state.
+	aclRole3 := mock.ACLRole()
+	aclRole3.ID = ""
+	aclRole3.Name = aclRole1.Name
+
+	aclRoleReq6 := &structs.ACLRolesUpsertRequest{
+		ACLRoles: []*structs.ACLRole{aclRole3},
+		WriteRequest: structs.WriteRequest{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp6 structs.ACLRolesUpsertResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLUpsertRolesRPCMethod, aclRoleReq6, &aclRoleResp6)
+	require.ErrorContains(t, err, fmt.Sprintf("role with name %s already exists", aclRole1.Name))
 }
 
 func TestACL_DeleteRolesByID(t *testing.T) {

nomad/state/state_store_acl.go

@@ -98,18 +98,52 @@ func (s *StateStore) upsertACLRoleTxn(
 		}
 	}
 
-	existing, err := txn.First(TableACLRoles, indexID, role.ID)
+	// This validation also happens within the RPC handler, but Raft latency
+	// could mean that by the time the state call is invoked, another Raft
+	// update has already written a role with the same name. We therefore need
+	// to check we are not trying to create a role with an existing name.
+	existingRaw, err := txn.First(TableACLRoles, indexName, role.Name)
 	if err != nil {
 		return false, fmt.Errorf("ACL role lookup failed: %v", err)
 	}
 
-	// Set up the indexes correctly to ensure existing indexes are maintained.
+	// Track our type asserted role, so we only need to do this once.
+	var existing *structs.ACLRole
+
+	// If we did not find an ACL Role within state with the same name, we need
+	// to check using the ID index as the operator might be performing an
+	// update on the role name.
+	//
+	// If we found an entry using the name index, we need to check that the ID
+	// matches the object within the request.
+	if existingRaw == nil {
+		existingRaw, err = txn.First(TableACLRoles, indexID, role.ID)
+		if err != nil {
+			return false, fmt.Errorf("ACL role lookup failed: %v", err)
+		}
+		if existingRaw != nil {
+			existing = existingRaw.(*structs.ACLRole)
+		}
+	} else {
+		existing = existingRaw.(*structs.ACLRole)
+		if existing.ID != role.ID {
+			return false, fmt.Errorf("ACL role with name %s already exists", role.Name)
+		}
+	}
+
+	// Depending on whether this is an initial create, or an update, we need to
+	// check and set certain parameters. The most important is to ensure any
+	// create index is carried over.
 	if existing != nil {
-		exist := existing.(*structs.ACLRole)
-		if exist.Equals(role) {
+
+		// If the role already exists, check whether the update contains any
+		// difference. If it doesn't, we can avoid a state update as wel as
+		// updates to any blocking queries.
+		if existing.Equals(role) {
 			return false, nil
 		}
-		role.CreateIndex = exist.CreateIndex
+
+		role.CreateIndex = existing.CreateIndex
 		role.ModifyIndex = index
 	} else {
 		role.CreateIndex = index

nomad/state/state_store_acl_test.go

@@ -1,6 +1,7 @@
 package state
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -246,6 +247,15 @@ func TestStateStore_UpsertACLRoles(t *testing.T) {
 	replicatedACLRoleResp, err := testState.GetACLRoleByName(ws, replicatedACLRole.Name)
 	require.NoError(t, err)
 	must.Eq(t, replicatedACLRole.Hash, replicatedACLRoleResp.Hash)
+
+	// Try adding a new ACL role, which has a name clash with an existing
+	// entry.
+	dupRoleName := mock.ACLRole()
+	dupRoleName.Name = mockedACLRoles[0].Name
+
+	err = testState.UpsertACLRoles(structs.MsgTypeTestSetup, 50,
+		[]*structs.ACLRole{dupRoleName}, false)
+	require.ErrorContains(t, err, fmt.Sprintf("ACL role with name %s already exists", dupRoleName.Name))
 }
 
 func TestStateStore_ValidateACLRolePolicyLinks(t *testing.T) {