rpc: add ACL Role RPC endpoint for CRUD actions.

New ACL Role RPC endpoints have been created to allow the creation, update, read, and deletion of ACL roles. All endpoints require a management token; in the future readers will also be allowed to view roles associated to their ACL token. The create endpoint in particular is responsible for deduplicating ACL policy links and ensuring named policies are found within state. This is done within the RPC handler so we perform a single loop through the links for slight efficiency.
2026-01-06 10:25:42 +03:00 · 2022-08-11 08:43:50 +01:00
parent d6a9c142ca
commit 4efb82af70
6 changed files with 912 additions and 8 deletions
--- a/nomad/acl_endpoint.go
+++ b/nomad/acl_endpoint.go
@@ -526,7 +526,7 @@ func (a *ACL) UpsertTokens(args *structs.ACLTokenUpsertRequest, reply *structs.A
 		if token.AccessorID != "" {
 			out, err := stateSnapshot.ACLTokenByAccessorID(nil, token.AccessorID)
 			if err != nil {
-				return structs.NewErrRPCCodedf(http.StatusBadRequest, "token lookup failed: %v", err)
+				return structs.NewErrRPCCodedf(http.StatusInternalServerError, "token lookup failed: %v", err)
 			}
 			if out == nil {
 				return structs.NewErrRPCCodedf(http.StatusBadRequest, "cannot find token %s", token.AccessorID)
@@ -1028,3 +1028,341 @@ func (a *ACL) ExpireOneTimeTokens(args *structs.OneTimeTokenExpireRequest, reply
 	reply.Index = index
 	return nil
 }
+
+// UpsertRoles creates or updates ACL roles held within Nomad.
+func (a *ACL) UpsertRoles(
+	args *structs.ACLRolesUpsertRequest,
+	reply *structs.ACLRolesUpsertResponse) error {
+
+	// Only allow operators to upsert ACL roles when ACLs are enabled.
+	if !a.srv.config.ACLEnabled {
+		return aclDisabled
+	}
+
+	// This endpoint always forwards to the authoritative region as ACL roles
+	// are global.
+	args.Region = a.srv.config.AuthoritativeRegion
+
+	if done, err := a.srv.forward(structs.ACLUpsertRolesRPCMethod, args, args, reply); done {
+		return err
+	}
+	defer metrics.MeasureSince([]string{"nomad", "acl", "upsert_roles"}, time.Now())
+
+	// Only tokens with management level permissions can create ACL roles.
+	if acl, err := a.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if acl == nil || !acl.IsManagement() {
+		return structs.ErrPermissionDenied
+	}
+
+	// Snapshot the state so we can perform lookups against the ID and policy
+	// links if needed. Do it here, so we only need to do this once no matter
+	// how many roles we are upserting.
+	stateSnapshot, err := a.srv.State().Snapshot()
+	if err != nil {
+		return err
+	}
+
+	// Validate each role.
+	for idx, role := range args.ACLRoles {
+
+		// Perform all the static validation of the ACL role object. Use the
+		// array index as we cannot be sure the error was caused by a missing
+		// name.
+		if err := role.Validate(); err != nil {
+			return structs.NewErrRPCCodedf(http.StatusBadRequest, "role %d invalid: %v", idx, err)
+		}
+
+		policyNames := make(map[string]struct{})
+		var policiesLinks []*structs.ACLRolePolicyLink
+
+		// We need to deduplicate the ACL policy links within this role as well
+		// as ensure the policies exist within state.
+		for _, policyLink := range role.Policies {
+
+			// Perform a state look up for the policy. An error or not being
+			// able to find the policy is terminal. We can include the name in
+			// the error message as it has previously been validated.
+			existing, err := stateSnapshot.ACLPolicyByName(nil, policyLink.Name)
+			if err != nil {
+				return structs.NewErrRPCCodedf(http.StatusInternalServerError, "policy lookup failed: %v", err)
+			}
+			if existing == nil {
+				return structs.NewErrRPCCodedf(http.StatusBadRequest, "cannot find policy %s", policyLink.Name)
+			}
+
+			// If the policy name is not found within our map, this means we
+			// have not seen it previously. We need to add this to our
+			// deduplicated array and also mark the policy name as seen, so we
+			// skip any future policies of the same name.
+			if _, ok := policyNames[policyLink.Name]; !ok {
+				policiesLinks = append(policiesLinks, policyLink)
+				policyNames[policyLink.Name] = struct{}{}
+			}
+		}
+
+		// 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()
+	}
+
+	// Update via Raft.
+	out, index, err := a.srv.raftApply(structs.ACLRolesUpsertRequestType, args)
+	if err != nil {
+		return err
+	}
+
+	// Check if the FSM response, which is an interface, contains an error.
+	if err, ok := out.(error); ok && err != nil {
+		return err
+	}
+
+	// Populate the response. We do a lookup against the state to pick up the
+	// proper create / modify times.
+	stateSnapshot, err = a.srv.State().Snapshot()
+	if err != nil {
+		return err
+	}
+	for _, role := range args.ACLRoles {
+		lookupACLRole, err := stateSnapshot.GetACLRoleByName(nil, role.Name)
+		if err != nil {
+			return structs.NewErrRPCCodedf(400, "ACL role lookup failed: %v", err)
+		}
+		reply.ACLRoles = append(reply.ACLRoles, lookupACLRole)
+	}
+
+	// Update the index. There is no need to floor this as we are writing to
+	// state and therefore will get a non-zero index response.
+	reply.Index = index
+	return nil
+}
+
+// DeleteRolesByID is used to batch delete ACL roles using the ID as the
+// deletion key.
+func (a *ACL) DeleteRolesByID(
+	args *structs.ACLRolesDeleteByIDRequest,
+	reply *structs.ACLRolesDeleteByIDResponse) error {
+
+	// Only allow operators to delete ACL roles when ACLs are enabled.
+	if !a.srv.config.ACLEnabled {
+		return aclDisabled
+	}
+
+	// This endpoint always forwards to the authoritative region as ACL roles
+	// are global.
+	args.Region = a.srv.config.AuthoritativeRegion
+
+	if done, err := a.srv.forward(structs.ACLDeleteRolesByIDRPCMethod, args, args, reply); done {
+		return err
+	}
+	defer metrics.MeasureSince([]string{"nomad", "acl", "delete_roles"}, time.Now())
+
+	// Only tokens with management level permissions can create ACL roles.
+	if acl, err := a.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if acl == nil || !acl.IsManagement() {
+		return structs.ErrPermissionDenied
+	}
+
+	// Update via Raft.
+	out, index, err := a.srv.raftApply(structs.ACLRolesDeleteByIDRequestType, args)
+	if err != nil {
+		return err
+	}
+
+	// Check if the FSM response, which is an interface, contains an error.
+	if err, ok := out.(error); ok && err != nil {
+		return err
+	}
+
+	// Update the index. There is no need to floor this as we are writing to
+	// state and therefore will get a non-zero index response.
+	reply.Index = index
+	return nil
+}
+
+// ListRoles is used to list ACL roles within state. If not prefix is supplied,
+// all ACL roles are listed, otherwise a prefix search is performed on the ACL
+// role name.
+func (a *ACL) ListRoles(
+	args *structs.ACLRolesListRequest,
+	reply *structs.ACLRolesListResponse) error {
+
+	// Only allow operators to list ACL roles when ACLs are enabled.
+	if !a.srv.config.ACLEnabled {
+		return aclDisabled
+	}
+
+	if done, err := a.srv.forward(structs.ACLListRolesRPCMethod, args, args, reply); done {
+		return err
+	}
+	defer metrics.MeasureSince([]string{"nomad", "acl", "list_roles"}, time.Now())
+
+	// TODO (jrasell) allow callers to list role associated to their token.
+	if acl, err := a.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if acl == nil || !acl.IsManagement() {
+		return structs.ErrPermissionDenied
+	}
+
+	// Set up and return the blocking query.
+	return a.srv.blockingRPC(&blockingOptions{
+		queryOpts: &args.QueryOptions,
+		queryMeta: &reply.QueryMeta,
+		run: func(ws memdb.WatchSet, stateStore *state.StateStore) error {
+
+			var (
+				err  error
+				iter memdb.ResultIterator
+			)
+
+			// If the operator supplied a prefix, perform a prefix search.
+			// Otherwise, list all ACL roles in state.
+			switch args.QueryOptions.Prefix {
+			case "":
+				iter, err = stateStore.GetACLRoles(ws)
+			default:
+				iter, err = stateStore.GetACLRoleByIDPrefix(ws, args.QueryOptions.Prefix)
+			}
+			if err != nil {
+				return err
+			}
+
+			// Iterate all the results and add these to our reply object. There
+			// is no stub object for an ACL role and the hash is needed by the
+			// replication process.
+			for raw := iter.Next(); raw != nil; raw = iter.Next() {
+				reply.ACLRoles = append(reply.ACLRoles, raw.(*structs.ACLRole))
+			}
+
+			// Use the index table to populate the query meta as we have no way
+			// of tracking the max index on deletes.
+			return a.srv.setReplyQueryMeta(stateStore, state.TableACLRoles, &reply.QueryMeta)
+		},
+	})
+}
+
+// GetRoleByID is used to look up an individual ACL role using its ID.
+func (a *ACL) GetRoleByID(
+	args *structs.ACLRoleByIDRequest,
+	reply *structs.ACLRoleByIDResponse) error {
+
+	// Only allow operators to read an ACL role when ACLs are enabled.
+	if !a.srv.config.ACLEnabled {
+		return aclDisabled
+	}
+
+	if done, err := a.srv.forward(structs.ACLGetRoleByIDRPCMethod, args, args, reply); done {
+		return err
+	}
+	defer metrics.MeasureSince([]string{"nomad", "acl", "get_role_id"}, time.Now())
+
+	// TODO (jrasell) allow callers to detail a role associated to their token.
+	if acl, err := a.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if acl == nil || !acl.IsManagement() {
+		return structs.ErrPermissionDenied
+	}
+
+	// Set up and return the blocking query.
+	return a.srv.blockingRPC(&blockingOptions{
+		queryOpts: &args.QueryOptions,
+		queryMeta: &reply.QueryMeta,
+		run: func(ws memdb.WatchSet, stateStore *state.StateStore) error {
+
+			// Perform a lookup for the ACL role.
+			out, err := stateStore.GetACLRoleByID(ws, args.RoleID)
+			if err != nil {
+				return err
+			}
+
+			// Set the index correctly depending on whether the ACL role was
+			// found.
+			switch out {
+			case nil:
+				index, err := stateStore.Index(state.TableACLRoles)
+				if err != nil {
+					return err
+				}
+				reply.Index = index
+			default:
+				reply.Index = out.ModifyIndex
+			}
+
+			// We didn't encounter an error looking up the index; set the ACL
+			// role on the reply and exit successfully.
+			reply.ACLRole = out
+			return nil
+		},
+	})
+}
+
+// GetRoleByName is used to look up an individual ACL role using its name.
+func (a *ACL) GetRoleByName(
+	args *structs.ACLRoleByNameRequest,
+	reply *structs.ACLRoleByNameResponse) error {
+
+	// Only allow operators to read an ACL role when ACLs are enabled.
+	if !a.srv.config.ACLEnabled {
+		return aclDisabled
+	}
+
+	if done, err := a.srv.forward(structs.ACLGetRoleByNameRPCMethod, args, args, reply); done {
+		return err
+	}
+	defer metrics.MeasureSince([]string{"nomad", "acl", "get_role_name"}, time.Now())
+
+	// TODO (jrasell) allow callers to detail a role associated to their token.
+	if acl, err := a.srv.ResolveToken(args.AuthToken); err != nil {
+		return err
+	} else if acl == nil || !acl.IsManagement() {
+		return structs.ErrPermissionDenied
+	}
+
+	// Set up and return the blocking query.
+	return a.srv.blockingRPC(&blockingOptions{
+		queryOpts: &args.QueryOptions,
+		queryMeta: &reply.QueryMeta,
+		run: func(ws memdb.WatchSet, stateStore *state.StateStore) error {
+
+			// Perform a lookup for the ACL role.
+			out, err := stateStore.GetACLRoleByName(ws, args.RoleName)
+			if err != nil {
+				return err
+			}
+
+			// Set the index correctly depending on whether the ACL role was
+			// found.
+			switch out {
+			case nil:
+				index, err := stateStore.Index(state.TableACLRoles)
+				if err != nil {
+					return err
+				}
+				reply.Index = index
+			default:
+				reply.Index = out.ModifyIndex
+			}
+
+			// We didn't encounter an error looking up the index; set the ACL
+			// role on the reply and exit successfully.
+			reply.ACLRole = out
+			return nil
+		},
+	})
+}
--- a/nomad/acl_endpoint_test.go
+++ b/nomad/acl_endpoint_test.go
@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"

+	"github.com/hashicorp/go-memdb"
 	msgpackrpc "github.com/hashicorp/net-rpc-msgpackrpc"
 	"github.com/hashicorp/nomad/ci"
 	"github.com/hashicorp/nomad/helper/uuid"
@@ -1776,3 +1777,389 @@ func TestACLEndpoint_OneTimeToken(t *testing.T) {
 	require.NoError(t, err)
 	require.Nil(t, ott)
 }
+
+func TestACL_UpsertRoles(t *testing.T) {
+	ci.Parallel(t)
+
+	testServer, aclRootToken, testServerCleanupFn := TestACLServer(t, nil)
+	defer testServerCleanupFn()
+	codec := rpcClient(t, testServer)
+	testutil.WaitForLeader(t, testServer.RPC)
+
+	// Create a mock ACL role and remove the ID so this looks like a creation.
+	aclRole1 := mock.ACLRole()
+	aclRole1.ID = ""
+
+	// Attempt to upsert this role without setting an ACL token. This should
+	// fail.
+	aclRoleReq1 := &structs.ACLRolesUpsertRequest{
+		ACLRoles: []*structs.ACLRole{aclRole1},
+		WriteRequest: structs.WriteRequest{
+			Region: "global",
+		},
+	}
+	var aclRoleResp1 structs.ACLRolesUpsertResponse
+	err := msgpackrpc.CallWithCodec(codec, structs.ACLUpsertRolesRPCMethod, aclRoleReq1, &aclRoleResp1)
+	require.ErrorContains(t, err, "Permission denied")
+
+	// Attempt to upsert this role again, this time setting the ACL root token.
+	// This should fail because the linked policies do not exist within state.
+	aclRoleReq2 := &structs.ACLRolesUpsertRequest{
+		ACLRoles: []*structs.ACLRole{aclRole1},
+		WriteRequest: structs.WriteRequest{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp2 structs.ACLRolesUpsertResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLUpsertRolesRPCMethod, aclRoleReq2, &aclRoleResp2)
+	require.ErrorContains(t, err, "cannot find policy")
+
+	// Create the policies our ACL roles wants to link to.
+	policy1 := mock.ACLPolicy()
+	policy1.Name = "mocked-test-policy-1"
+	policy2 := mock.ACLPolicy()
+	policy2.Name = "mocked-test-policy-2"
+
+	require.NoError(t, testServer.fsm.State().UpsertACLPolicies(
+		structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+	// Try the upsert a third time, which should succeed.
+	aclRoleReq3 := &structs.ACLRolesUpsertRequest{
+		ACLRoles: []*structs.ACLRole{aclRole1},
+		WriteRequest: structs.WriteRequest{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp3 structs.ACLRolesUpsertResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLUpsertRolesRPCMethod, aclRoleReq3, &aclRoleResp3)
+	require.NoError(t, err)
+	require.Len(t, aclRoleResp3.ACLRoles, 1)
+	require.True(t, aclRole1.Equals(aclRoleResp3.ACLRoles[0]))
+
+	// Perform an update of the ACL role by removing a policy and changing the
+	// name.
+	aclRole1Copy := aclRole1.Copy()
+	aclRole1Copy.Name = "updated-role-name"
+	aclRole1Copy.Policies = append(aclRole1Copy.Policies[:1], aclRole1Copy.Policies[1+1:]...)
+	aclRole1Copy.SetHash()
+
+	aclRoleReq4 := &structs.ACLRolesUpsertRequest{
+		ACLRoles: []*structs.ACLRole{aclRole1Copy},
+		WriteRequest: structs.WriteRequest{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp4 structs.ACLRolesUpsertResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLUpsertRolesRPCMethod, aclRoleReq4, &aclRoleResp4)
+	require.NoError(t, err)
+	require.Len(t, aclRoleResp4.ACLRoles, 1)
+	require.True(t, aclRole1Copy.Equals(aclRoleResp4.ACLRoles[0]))
+	require.Greater(t, aclRoleResp4.ACLRoles[0].ModifyIndex, aclRoleResp3.ACLRoles[0].ModifyIndex)
+
+	// Create another ACL role that will fail validation. Attempting to upsert
+	// this ensures the handler is triggering the validation function.
+	aclRole2 := mock.ACLRole()
+	aclRole2.Policies = nil
+
+	aclRoleReq5 := &structs.ACLRolesUpsertRequest{
+		ACLRoles: []*structs.ACLRole{aclRole2},
+		WriteRequest: structs.WriteRequest{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp5 structs.ACLRolesUpsertResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLUpsertRolesRPCMethod, aclRoleReq5, &aclRoleResp5)
+	require.Error(t, err)
+	require.NotContains(t, err, "Permission denied")
+}
+
+func TestACL_DeleteRolesByID(t *testing.T) {
+	ci.Parallel(t)
+
+	testServer, aclRootToken, testServerCleanupFn := TestACLServer(t, nil)
+	defer testServerCleanupFn()
+	codec := rpcClient(t, testServer)
+	testutil.WaitForLeader(t, testServer.RPC)
+
+	// Create the policies our ACL roles wants to link to.
+	policy1 := mock.ACLPolicy()
+	policy1.Name = "mocked-test-policy-1"
+	policy2 := mock.ACLPolicy()
+	policy2.Name = "mocked-test-policy-2"
+
+	require.NoError(t, testServer.fsm.State().UpsertACLPolicies(
+		structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+	// Create two ACL roles and put these directly into state.
+	aclRoles := []*structs.ACLRole{mock.ACLRole(), mock.ACLRole()}
+	require.NoError(t, testServer.State().UpsertACLRoles(structs.MsgTypeTestSetup, 10, aclRoles))
+
+	// Attempt to delete an ACL role without setting an auth token. This should
+	// fail.
+	aclRoleReq1 := &structs.ACLRolesDeleteByIDRequest{
+		ACLRoleIDs: []string{aclRoles[0].ID},
+		WriteRequest: structs.WriteRequest{
+			Region: DefaultRegion,
+		},
+	}
+	var aclRoleResp1 structs.ACLRolesDeleteByIDResponse
+	err := msgpackrpc.CallWithCodec(codec, structs.ACLDeleteRolesByIDRPCMethod, aclRoleReq1, &aclRoleResp1)
+	require.ErrorContains(t, err, "Permission denied")
+
+	// Attempt to delete an ACL role now using a valid management token which
+	// should succeed.
+	aclRoleReq2 := &structs.ACLRolesDeleteByIDRequest{
+		ACLRoleIDs: []string{aclRoles[0].ID},
+		WriteRequest: structs.WriteRequest{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp2 structs.ACLRolesDeleteByIDResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLDeleteRolesByIDRPCMethod, aclRoleReq2, &aclRoleResp2)
+	require.NoError(t, err)
+
+	// Ensure the deleted role is not found within state and that the other is.
+	ws := memdb.NewWatchSet()
+	iter, err := testServer.State().GetACLRoles(ws)
+	require.NoError(t, err)
+
+	var aclRolesLookup []*structs.ACLRole
+	for raw := iter.Next(); raw != nil; raw = iter.Next() {
+		aclRolesLookup = append(aclRolesLookup, raw.(*structs.ACLRole))
+	}
+
+	require.Len(t, aclRolesLookup, 1)
+	require.True(t, aclRolesLookup[0].Equals(aclRoles[1]))
+
+	// Try to delete the previously deleted ACL role, this should fail.
+	aclRoleReq3 := &structs.ACLRolesDeleteByIDRequest{
+		ACLRoleIDs: []string{aclRoles[0].ID},
+		WriteRequest: structs.WriteRequest{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp3 structs.ACLRolesDeleteByIDResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLDeleteRolesByIDRPCMethod, aclRoleReq3, &aclRoleResp3)
+	require.ErrorContains(t, err, "ACL role not found")
+}
+
+func TestACL_ListRoles(t *testing.T) {
+	ci.Parallel(t)
+
+	testServer, aclRootToken, testServerCleanupFn := TestACLServer(t, nil)
+	defer testServerCleanupFn()
+	codec := rpcClient(t, testServer)
+	testutil.WaitForLeader(t, testServer.RPC)
+
+	// Create the policies our ACL roles wants to link to.
+	policy1 := mock.ACLPolicy()
+	policy1.Name = "mocked-test-policy-1"
+	policy2 := mock.ACLPolicy()
+	policy2.Name = "mocked-test-policy-2"
+
+	require.NoError(t, testServer.fsm.State().UpsertACLPolicies(
+		structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+	// Create two ACL roles with a known prefix and put these directly into
+	// state.
+	aclRoles := []*structs.ACLRole{mock.ACLRole(), mock.ACLRole()}
+	aclRoles[0].ID = "prefix-" + uuid.Generate()
+	aclRoles[1].ID = "prefix-" + uuid.Generate()
+	require.NoError(t, testServer.State().UpsertACLRoles(structs.MsgTypeTestSetup, 10, aclRoles))
+
+	// Try listing roles without a valid ACL token.
+	aclRoleReq1 := &structs.ACLRolesListRequest{
+		QueryOptions: structs.QueryOptions{
+			Region: DefaultRegion,
+		},
+	}
+	var aclRoleResp1 structs.ACLRolesListResponse
+	err := msgpackrpc.CallWithCodec(codec, structs.ACLListRolesRPCMethod, aclRoleReq1, &aclRoleResp1)
+	require.ErrorContains(t, err, "Permission denied")
+
+	// Try listing roles with a valid ACL token.
+	aclRoleReq2 := &structs.ACLRolesListRequest{
+		QueryOptions: structs.QueryOptions{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp2 structs.ACLRolesListResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLListRolesRPCMethod, aclRoleReq2, &aclRoleResp2)
+	require.NoError(t, err)
+	require.Len(t, aclRoleResp2.ACLRoles, 2)
+
+	// Try listing roles with a valid ACL token using a prefix that doesn't
+	// match anything.
+	aclRoleReq3 := &structs.ACLRolesListRequest{
+		QueryOptions: structs.QueryOptions{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+			Prefix:    "please",
+		},
+	}
+	var aclRoleResp3 structs.ACLRolesListResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLListRolesRPCMethod, aclRoleReq3, &aclRoleResp3)
+	require.NoError(t, err)
+	require.Len(t, aclRoleResp3.ACLRoles, 0)
+
+	// Try listing roles with a valid ACL token using a prefix that matches two
+	// entries.
+	aclRoleReq4 := &structs.ACLRolesListRequest{
+		QueryOptions: structs.QueryOptions{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+			Prefix:    "prefix-",
+		},
+	}
+	var aclRoleResp4 structs.ACLRolesListResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLListRolesRPCMethod, aclRoleReq4, &aclRoleResp4)
+	require.NoError(t, err)
+	require.Len(t, aclRoleResp4.ACLRoles, 2)
+}
+
+func TestACL_GetRoleByID(t *testing.T) {
+	ci.Parallel(t)
+
+	testServer, aclRootToken, testServerCleanupFn := TestACLServer(t, nil)
+	defer testServerCleanupFn()
+	codec := rpcClient(t, testServer)
+	testutil.WaitForLeader(t, testServer.RPC)
+
+	// Create the policies our ACL roles wants to link to.
+	policy1 := mock.ACLPolicy()
+	policy1.Name = "mocked-test-policy-1"
+	policy2 := mock.ACLPolicy()
+	policy2.Name = "mocked-test-policy-2"
+
+	require.NoError(t, testServer.fsm.State().UpsertACLPolicies(
+		structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+	// Create two ACL roles and put these directly into state.
+	aclRoles := []*structs.ACLRole{mock.ACLRole(), mock.ACLRole()}
+	require.NoError(t, testServer.State().UpsertACLRoles(structs.MsgTypeTestSetup, 10, aclRoles))
+
+	// Try reading a role without setting a correct auth token.
+	aclRoleReq1 := &structs.ACLRoleByIDRequest{
+		QueryOptions: structs.QueryOptions{
+			Region: DefaultRegion,
+		},
+	}
+	var aclRoleResp1 structs.ACLRoleByIDResponse
+	err := msgpackrpc.CallWithCodec(codec, structs.ACLGetRoleByIDRPCMethod, aclRoleReq1, &aclRoleResp1)
+	require.ErrorContains(t, err, "Permission denied")
+
+	// Try reading a role that doesn't exist.
+	aclRoleReq2 := &structs.ACLRoleByIDRequest{
+		RoleID: "nope",
+		QueryOptions: structs.QueryOptions{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp2 structs.ACLRoleByIDResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLGetRoleByIDRPCMethod, aclRoleReq2, &aclRoleResp2)
+	require.NoError(t, err)
+	require.Nil(t, aclRoleResp2.ACLRole)
+
+	// Read both our available ACL roles using a valid auth token.
+	aclRoleReq3 := &structs.ACLRoleByIDRequest{
+		RoleID: aclRoles[0].ID,
+		QueryOptions: structs.QueryOptions{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp3 structs.ACLRoleByIDResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLGetRoleByIDRPCMethod, aclRoleReq3, &aclRoleResp3)
+	require.NoError(t, err)
+	require.True(t, aclRoleResp3.ACLRole.Equals(aclRoles[0]))
+
+	aclRoleReq4 := &structs.ACLRoleByIDRequest{
+		RoleID: aclRoles[1].ID,
+		QueryOptions: structs.QueryOptions{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp4 structs.ACLRoleByIDResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLGetRoleByIDRPCMethod, aclRoleReq4, &aclRoleResp4)
+	require.NoError(t, err)
+	require.True(t, aclRoleResp4.ACLRole.Equals(aclRoles[1]))
+}
+
+func TestACL_GetRoleByName(t *testing.T) {
+	ci.Parallel(t)
+
+	testServer, aclRootToken, testServerCleanupFn := TestACLServer(t, nil)
+	defer testServerCleanupFn()
+	codec := rpcClient(t, testServer)
+	testutil.WaitForLeader(t, testServer.RPC)
+
+	// Create the policies our ACL roles wants to link to.
+	policy1 := mock.ACLPolicy()
+	policy1.Name = "mocked-test-policy-1"
+	policy2 := mock.ACLPolicy()
+	policy2.Name = "mocked-test-policy-2"
+
+	require.NoError(t, testServer.fsm.State().UpsertACLPolicies(
+		structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+	// Create two ACL roles and put these directly into state.
+	aclRoles := []*structs.ACLRole{mock.ACLRole(), mock.ACLRole()}
+	require.NoError(t, testServer.State().UpsertACLRoles(structs.MsgTypeTestSetup, 10, aclRoles))
+
+	// Try reading a role without setting a correct auth token.
+	aclRoleReq1 := &structs.ACLRoleByNameRequest{
+		QueryOptions: structs.QueryOptions{
+			Region: DefaultRegion,
+		},
+	}
+	var aclRoleResp1 structs.ACLRoleByNameResponse
+	err := msgpackrpc.CallWithCodec(codec, structs.ACLGetRoleByNameRPCMethod, aclRoleReq1, &aclRoleResp1)
+	require.ErrorContains(t, err, "Permission denied")
+
+	// Try reading a role that doesn't exist.
+	aclRoleReq2 := &structs.ACLRoleByNameRequest{
+		RoleName: "nope",
+		QueryOptions: structs.QueryOptions{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp2 structs.ACLRoleByNameResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLGetRoleByNameRPCMethod, aclRoleReq2, &aclRoleResp2)
+	require.NoError(t, err)
+	require.Nil(t, aclRoleResp2.ACLRole)
+
+	// Read both our available ACL roles using a valid auth token.
+	aclRoleReq3 := &structs.ACLRoleByNameRequest{
+		RoleName: aclRoles[0].Name,
+		QueryOptions: structs.QueryOptions{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp3 structs.ACLRoleByNameResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLGetRoleByNameRPCMethod, aclRoleReq3, &aclRoleResp3)
+	require.NoError(t, err)
+	require.True(t, aclRoleResp3.ACLRole.Equals(aclRoles[0]))
+
+	aclRoleReq4 := &structs.ACLRoleByNameRequest{
+		RoleName: aclRoles[1].Name,
+		QueryOptions: structs.QueryOptions{
+			Region:    DefaultRegion,
+			AuthToken: aclRootToken.SecretID,
+		},
+	}
+	var aclRoleResp4 structs.ACLRoleByNameResponse
+	err = msgpackrpc.CallWithCodec(codec, structs.ACLGetRoleByNameRPCMethod, aclRoleReq4, &aclRoleResp4)
+	require.NoError(t, err)
+	require.True(t, aclRoleResp4.ACLRole.Equals(aclRoles[1]))
+}
--- a/nomad/state/state_store_acl.go
+++ b/nomad/state/state_store_acl.go
@@ -120,13 +120,6 @@ func (s *StateStore) upsertACLRoleTxn(
 	return true, nil
 }

-// ValidateACLRolePolicyLinks ensures all ACL policies linked to from the ACL
-// role exist within state.
-func (s *StateStore) ValidateACLRolePolicyLinks(role *structs.ACLRole) error {
-	txn := s.db.ReadTxn()
-	return s.validateACLRolePolicyLinksTxn(txn, role)
-}
-
 // validateACLRolePolicyLinksTxn is the same as ValidateACLRolePolicyLinks but
 // allows callers to pass their own transaction.
 func (s *StateStore) validateACLRolePolicyLinksTxn(txn *txn, role *structs.ACLRole) error {
@@ -228,3 +221,17 @@ func (s *StateStore) GetACLRoleByName(ws memdb.WatchSet, roleName string) (*stru
 	}
 	return nil, nil
 }
+
+// GetACLRoleByIDPrefix is used to lookup ACL policies using a prefix to match
+// on the ID.
+func (s *StateStore) GetACLRoleByIDPrefix(ws memdb.WatchSet, idPrefix string) (memdb.ResultIterator, error) {
+	txn := s.db.ReadTxn()
+
+	iter, err := txn.Get(TableACLRoles, indexID+"_prefix", idPrefix)
+	if err != nil {
+		return nil, fmt.Errorf("ACL role lookup failed: %v", err)
+	}
+	ws.Add(iter.WatchCh())
+
+	return iter, nil
+}
--- a/nomad/state/state_store_acl_test.go
+++ b/nomad/state/state_store_acl_test.go
@@ -7,6 +7,7 @@ import (
 	"github.com/hashicorp/go-memdb"
 	"github.com/hashicorp/nomad/ci"
 	"github.com/hashicorp/nomad/helper/pointer"
+	"github.com/hashicorp/nomad/helper/uuid"
 	"github.com/hashicorp/nomad/nomad/mock"
 	"github.com/hashicorp/nomad/nomad/structs"
 	"github.com/shoenig/test/must"
@@ -443,3 +444,47 @@ func TestStateStore_GetACLRoleByName(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, mockedACLRoles[1], aclRole)
 }
+
+func TestStateStore_GetACLRoleByIDPrefix(t *testing.T) {
+	ci.Parallel(t)
+	testState := testStateStore(t)
+
+	// Create the policies our ACL roles wants to link to.
+	policy1 := mock.ACLPolicy()
+	policy1.Name = "mocked-test-policy-1"
+	policy2 := mock.ACLPolicy()
+	policy2.Name = "mocked-test-policy-2"
+
+	require.NoError(t, testState.UpsertACLPolicies(
+		structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+	// Generate a some mocked ACL roles for testing and upsert these straight
+	// into state. Set the ID to something with a prefix we know so it is easy
+	// to test.
+	mockedACLRoles := []*structs.ACLRole{mock.ACLRole(), mock.ACLRole()}
+	mockedACLRoles[0].ID = "test-prefix-" + uuid.Generate()
+	mockedACLRoles[1].ID = "test-prefix-" + uuid.Generate()
+	require.NoError(t, testState.UpsertACLRoles(structs.MsgTypeTestSetup, 10, mockedACLRoles))
+
+	ws := memdb.NewWatchSet()
+
+	// Try using a prefix that doesn't match any entries.
+	iter, err := testState.GetACLRoleByIDPrefix(ws, "nope")
+	require.NoError(t, err)
+
+	var aclRoles []*structs.ACLRole
+	for raw := iter.Next(); raw != nil; raw = iter.Next() {
+		aclRoles = append(aclRoles, raw.(*structs.ACLRole))
+	}
+	require.Len(t, aclRoles, 0)
+
+	// Use a prefix which should match two entries in state.
+	iter, err = testState.GetACLRoleByIDPrefix(ws, "test-prefix-")
+	require.NoError(t, err)
+
+	aclRoles = []*structs.ACLRole{}
+	for raw := iter.Next(); raw != nil; raw = iter.Next() {
+		aclRoles = append(aclRoles, raw.(*structs.ACLRole))
+	}
+	require.Len(t, aclRoles, 2)
+}
--- a/nomad/structs/acl.go
+++ b/nomad/structs/acl.go
@@ -35,6 +35,40 @@ const (
 	// Args: ACLTokenDeleteRequest
 	// Reply: GenericResponse
 	ACLDeleteTokensRPCMethod = "ACL.DeleteTokens"
+
+	// ACLUpsertRolesRPCMethod is the RPC method for batch creating or
+	// modifying ACL roles.
+	//
+	// Args: ACLRolesUpsertRequest
+	// Reply: ACLRolesUpsertResponse
+	ACLUpsertRolesRPCMethod = "ACL.UpsertRoles"
+
+	// ACLDeleteRolesByIDRPCMethod the RPC method for batch deleting ACL
+	// roles by their ID.
+	//
+	// Args: ACLRolesDeleteByIDRequest
+	// Reply: ACLRolesDeleteByIDResponse
+	ACLDeleteRolesByIDRPCMethod = "ACL.DeleteRolesByID"
+
+	// ACLListRolesRPCMethod is the RPC method for listing ACL roles.
+	//
+	// Args: ACLRolesListRequest
+	// Reply: ACLRolesListResponse
+	ACLListRolesRPCMethod = "ACL.ListRoles"
+
+	// ACLGetRoleByIDRPCMethod is the RPC method for detailing an individual
+	// ACL role using its ID.
+	//
+	// Args: ACLRoleByIDRequest
+	// Reply: ACLRoleByIDResponse
+	ACLGetRoleByIDRPCMethod = "ACL.GetRoleByID"
+
+	// ACLGetRoleByNameRPCMethod is the RPC method for detailing an individual
+	// ACL role using its name.
+	//
+	// Args: ACLRoleByNameRequest
+	// Reply: ACLRoleByNameResponse
+	ACLGetRoleByNameRPCMethod = "ACL.GetRoleByName"
 )

 const (
@@ -266,6 +300,15 @@ func (a *ACLRole) Validate() error {
 	return mErr.ErrorOrNil()
 }

+// Canonicalize performs basic canonicalization on the ACL role object. It is
+// important for callers to understand certain fields such as ID are set if it
+// is empty, so copies should be taken if needed before calling this function.
+func (a *ACLRole) Canonicalize() {
+	if a.ID == "" {
+		a.ID = uuid.Generate()
+	}
+}
+
 // Equals performs an equality check on the two service registrations. It
 // handles nil objects.
 func (a *ACLRole) Equals(o *ACLRole) bool {
@@ -307,6 +350,7 @@ type ACLRolesUpsertRequest struct {
 // ACLRolesUpsertResponse is the response object when one or more ACL roles
 // have been successfully upserted into state.
 type ACLRolesUpsertResponse struct {
+	ACLRoles []*ACLRole
 	WriteMeta
 }

@@ -322,3 +366,43 @@ type ACLRolesDeleteByIDRequest struct {
 type ACLRolesDeleteByIDResponse struct {
 	WriteMeta
 }
+
+// ACLRolesListRequest is the request object when performing ACL role listings.
+type ACLRolesListRequest struct {
+	QueryOptions
+}
+
+// ACLRolesListResponse is the response object when performing ACL role
+// listings.
+type ACLRolesListResponse struct {
+	ACLRoles []*ACLRole
+	QueryMeta
+}
+
+// ACLRoleByIDRequest is the request object to perform a lookup of an ACL
+// role using a specific ID.
+type ACLRoleByIDRequest struct {
+	RoleID string
+	QueryOptions
+}
+
+// ACLRoleByIDResponse is the response object when performing a lookup of an
+// ACL role matching a specific ID.
+type ACLRoleByIDResponse struct {
+	ACLRole *ACLRole
+	QueryMeta
+}
+
+// ACLRoleByNameRequest is the request object to perform a lookup of an ACL
+// role using a specific name.
+type ACLRoleByNameRequest struct {
+	RoleName string
+	QueryOptions
+}
+
+// ACLRoleByNameResponse is the response object when performing a lookup of an
+// ACL role matching a specific name.
+type ACLRoleByNameResponse struct {
+	ACLRole *ACLRole
+	QueryMeta
+}
--- a/nomad/structs/acl_test.go
+++ b/nomad/structs/acl_test.go
@@ -418,6 +418,34 @@ func TestACLRole_Validate(t *testing.T) {
 	}
 }

+func TestACLRole_Canonicalize(t *testing.T) {
+	testCases := []struct {
+		name         string
+		inputACLRole *ACLRole
+	}{
+		{
+			name:         "no ID set",
+			inputACLRole: &ACLRole{},
+		},
+		{
+			name:         "id set",
+			inputACLRole: &ACLRole{ID: "some-random-uuid"},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			existing := tc.inputACLRole.Copy()
+			tc.inputACLRole.Canonicalize()
+			if existing.ID == "" {
+				require.NotEmpty(t, tc.inputACLRole.ID)
+			} else {
+				require.Equal(t, existing.ID, tc.inputACLRole.ID)
+			}
+		})
+	}
+}
+
 func TestACLRole_Equals(t *testing.T) {
 	testCases := []struct {
 		name            string
@@ -638,3 +666,18 @@ func Test_ACLRolesDeleteByIDRequest(t *testing.T) {
 	req := ACLRolesDeleteByIDRequest{}
 	require.False(t, req.IsRead())
 }
+
+func Test_ACLRolesListRequest(t *testing.T) {
+	req := ACLRolesListRequest{}
+	require.True(t, req.IsRead())
+}
+
+func Test_ACLRoleByIDRequest(t *testing.T) {
+	req := ACLRoleByIDRequest{}
+	require.True(t, req.IsRead())
+}
+
+func Test_ACLRoleByNameRequest(t *testing.T) {
+	req := ACLRoleByNameRequest{}
+	require.True(t, req.IsRead())
+}