diff --git a/api/acl.go b/api/acl.go
index 07b51deb4..fd7f20786 100644
--- a/api/acl.go
+++ b/api/acl.go
@@ -1,6 +1,7 @@
 package api
 
 import (
+	"errors"
 	"fmt"
 	"time"
 )
@@ -202,6 +203,96 @@ func (a *ACLTokens) ExchangeOneTimeToken(secret string, q *WriteOptions) (*ACLTo
 	return resp.Token, wm, nil
 }
 
+var (
+	// errMissingACLRoleID is the generic errors to use when a call is missing
+	// the required ACL Role ID parameter.
+	errMissingACLRoleID = errors.New("missing ACL role ID")
+)
+
+// ACLRoles is used to query the ACL Role endpoints.
+type ACLRoles struct {
+	client *Client
+}
+
+// ACLRoles returns a new handle on the ACL roles API client.
+func (c *Client) ACLRoles() *ACLRoles {
+	return &ACLRoles{client: c}
+}
+
+// List is used to detail all the ACL roles currently stored within state.
+func (a *ACLRoles) List(q *QueryOptions) ([]*ACLRole, *QueryMeta, error) {
+	var resp []*ACLRole
+	qm, err := a.client.query("/v1/acl/roles", &resp, q)
+	if err != nil {
+		return nil, nil, err
+	}
+	return resp, qm, nil
+}
+
+// Create is used to create an ACL role.
+func (a *ACLRoles) Create(role *ACLRole, w *WriteOptions) (*ACLRole, *WriteMeta, error) {
+	if role.ID != "" {
+		return nil, nil, errors.New("cannot specify ACL role ID")
+	}
+	var resp ACLRole
+	wm, err := a.client.write("/v1/acl/role", role, &resp, w)
+	if err != nil {
+		return nil, nil, err
+	}
+	return &resp, wm, nil
+}
+
+// Update is used to update an existing ACL role.
+func (a *ACLRoles) Update(role *ACLRole, w *WriteOptions) (*ACLRole, *WriteMeta, error) {
+	if role.ID == "" {
+		return nil, nil, errMissingACLRoleID
+	}
+	var resp ACLRole
+	wm, err := a.client.write("/v1/acl/role/"+role.ID, role, &resp, w)
+	if err != nil {
+		return nil, nil, err
+	}
+	return &resp, wm, nil
+}
+
+// Delete is used to delete an ACL role.
+func (a *ACLRoles) Delete(roleID string, w *WriteOptions) (*WriteMeta, error) {
+	if roleID == "" {
+		return nil, errMissingACLRoleID
+	}
+	wm, err := a.client.delete("/v1/acl/role/"+roleID, nil, nil, w)
+	if err != nil {
+		return nil, err
+	}
+	return wm, nil
+}
+
+// Get is used to look up an ACL role.
+func (a *ACLRoles) Get(roleID string, q *QueryOptions) (*ACLRole, *QueryMeta, error) {
+	if roleID == "" {
+		return nil, nil, errMissingACLRoleID
+	}
+	var resp ACLRole
+	qm, err := a.client.query("/v1/acl/role/"+roleID, &resp, q)
+	if err != nil {
+		return nil, nil, err
+	}
+	return &resp, qm, nil
+}
+
+// GetByName is used to look up an ACL role using its name.
+func (a *ACLRoles) GetByName(roleName string, q *QueryOptions) (*ACLRole, *QueryMeta, error) {
+	if roleName == "" {
+		return nil, nil, errors.New("missing ACL role name")
+	}
+	var resp ACLRole
+	qm, err := a.client.query("/v1/acl/role/name/"+roleName, &resp, q)
+	if err != nil {
+		return nil, nil, err
+	}
+	return &resp, qm, nil
+}
+
 // ACLPolicyListStub is used to for listing ACL policies
 type ACLPolicyListStub struct {
 	Name        string
@@ -285,3 +376,42 @@ type OneTimeTokenExchangeResponse struct {
 type BootstrapRequest struct {
 	BootstrapSecret string
 }
+
+// ACLRole is an abstraction for the ACL system which allows the grouping of
+// ACL policies into a single object. ACL tokens can be created and linked to
+// a role; the token then inherits all the permissions granted by the policies.
+type ACLRole struct {
+
+	// ID is an internally generated UUID for this role and is controlled by
+	// Nomad. It can be used after role creation to update the existing role.
+	ID string
+
+	// Name is unique across the entire set of federated clusters and is
+	// supplied by the operator on role creation. The name can be modified by
+	// updating the role and including the Nomad generated ID. This update will
+	// not affect tokens created and linked to this role. This is a required
+	// field.
+	Name string
+
+	// Description is a human-readable, operator set description that can
+	// provide additional context about the role. This is an optional field.
+	Description string
+
+	// Policies is an array of ACL policy links. Although currently policies
+	// can only be linked using their name, in the future we will want to add
+	// IDs also and thus allow operators to specify either a name, an ID, or
+	// both. At least one entry is required.
+	Policies []*ACLRolePolicyLink
+
+	CreateIndex uint64
+	ModifyIndex uint64
+}
+
+// ACLRolePolicyLink is used to link a policy to an ACL role. We use a struct
+// rather than a list of strings as in the future we will want to add IDs to
+// policies and then link via these.
+type ACLRolePolicyLink struct {
+
+	// Name is the ACLPolicy.Name value which will be linked to the ACL role.
+	Name string
+}
diff --git a/api/acl_test.go b/api/acl_test.go
index 7a0434b54..4487a7778 100644
--- a/api/acl_test.go
+++ b/api/acl_test.go
@@ -346,3 +346,77 @@ func TestACLTokens_BootstrapValidToken(t *testing.T) {
 	assertWriteMeta(t, wm)
 	assert.Equal(t, bootkn, out.SecretID)
 }
+
+func TestACLRoles(t *testing.T) {
+	testutil.Parallel(t)
+
+	testClient, testServer, _ := makeACLClient(t, nil, nil)
+	defer testServer.Stop()
+
+	// An initial listing shouldn't return any results.
+	aclRoleListResp, queryMeta, err := testClient.ACLRoles().List(nil)
+	require.NoError(t, err)
+	require.Empty(t, aclRoleListResp)
+	assertQueryMeta(t, queryMeta)
+
+	// Create an ACL policy that can be referenced within the ACL role.
+	aclPolicy := ACLPolicy{
+		Name: "acl-role-api-test",
+		Rules: `namespace "default" {
+			policy = "read"
+		}
+		`,
+	}
+	writeMeta, err := testClient.ACLPolicies().Upsert(&aclPolicy, nil)
+	require.NoError(t, err)
+	assertWriteMeta(t, writeMeta)
+
+	// Create an ACL role referencing the previously created policy.
+	role := ACLRole{
+		Name:     "acl-role-api-test",
+		Policies: []*ACLRolePolicyLink{{Name: aclPolicy.Name}},
+	}
+	aclRoleCreateResp, writeMeta, err := testClient.ACLRoles().Create(&role, nil)
+	require.NoError(t, err)
+	assertWriteMeta(t, writeMeta)
+	require.NotEmpty(t, aclRoleCreateResp.ID)
+	require.Equal(t, role.Name, aclRoleCreateResp.Name)
+
+	// Another listing should return one result.
+	aclRoleListResp, queryMeta, err = testClient.ACLRoles().List(nil)
+	require.NoError(t, err)
+	require.Len(t, aclRoleListResp, 1)
+	assertQueryMeta(t, queryMeta)
+
+	// Read the role using its ID.
+	aclRoleReadResp, queryMeta, err := testClient.ACLRoles().Get(aclRoleCreateResp.ID, nil)
+	require.NoError(t, err)
+	assertQueryMeta(t, queryMeta)
+	require.Equal(t, aclRoleCreateResp, aclRoleReadResp)
+
+	// Read the role using its name.
+	aclRoleReadResp, queryMeta, err = testClient.ACLRoles().GetByName(aclRoleCreateResp.Name, nil)
+	require.NoError(t, err)
+	assertQueryMeta(t, queryMeta)
+	require.Equal(t, aclRoleCreateResp, aclRoleReadResp)
+
+	// Update the role name.
+	role.Name = "acl-role-api-test-badger-badger-badger"
+	role.ID = aclRoleCreateResp.ID
+	aclRoleUpdateResp, writeMeta, err := testClient.ACLRoles().Update(&role, nil)
+	require.NoError(t, err)
+	assertWriteMeta(t, writeMeta)
+	require.Equal(t, role.Name, aclRoleUpdateResp.Name)
+	require.Equal(t, role.ID, aclRoleUpdateResp.ID)
+
+	// Delete the role.
+	writeMeta, err = testClient.ACLRoles().Delete(aclRoleCreateResp.ID, nil)
+	require.NoError(t, err)
+	assertWriteMeta(t, writeMeta)
+
+	// Make sure there are no ACL roles now present.
+	aclRoleListResp, queryMeta, err = testClient.ACLRoles().List(nil)
+	require.NoError(t, err)
+	require.Empty(t, aclRoleListResp)
+	assertQueryMeta(t, queryMeta)
+}
diff --git a/command/agent/acl_endpoint.go b/command/agent/acl_endpoint.go
index df3f56851..ceaecb0b8 100644
--- a/command/agent/acl_endpoint.go
+++ b/command/agent/acl_endpoint.go
@@ -322,3 +322,206 @@ func (s *HTTPServer) ExchangeOneTimeToken(resp http.ResponseWriter, req *http.Re
 	setIndex(resp, out.Index)
 	return out, nil
 }
+
+// ACLRoleListRequest performs a listing of ACL roles and is callable via the
+// /v1/acl/roles HTTP API.
+func (s *HTTPServer) ACLRoleListRequest(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
+
+	// The endpoint only supports GET requests.
+	if req.Method != http.MethodGet {
+		return nil, CodedError(http.StatusMethodNotAllowed, ErrInvalidMethod)
+	}
+
+	// Set up the request args and parse this to ensure the query options are
+	// set.
+	args := structs.ACLRolesListRequest{}
+
+	if s.parse(resp, req, &args.Region, &args.QueryOptions) {
+		return nil, nil
+	}
+
+	// Perform the RPC request.
+	var reply structs.ACLRolesListResponse
+	if err := s.agent.RPC(structs.ACLListRolesRPCMethod, &args, &reply); err != nil {
+		return nil, err
+	}
+
+	setMeta(resp, &reply.QueryMeta)
+
+	if reply.ACLRoles == nil {
+		reply.ACLRoles = make([]*structs.ACLRole, 0)
+	}
+	return reply.ACLRoles, nil
+}
+
+// ACLRoleRequest creates a new ACL role and is callable via the
+// /v1/acl/role HTTP API.
+func (s *HTTPServer) ACLRoleRequest(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
+
+	// // The endpoint only supports PUT or POST requests.
+	if !(req.Method == http.MethodPut || req.Method == http.MethodPost) {
+		return nil, CodedError(http.StatusMethodNotAllowed, ErrInvalidMethod)
+	}
+
+	// Use the generic upsert function without setting an ID as this will be
+	// handled by the Nomad leader.
+	return s.aclRoleUpsertRequest(resp, req, "")
+}
+
+// ACLRoleSpecificRequest is callable via the /v1/acl/role/ HTTP API and
+// handles read via both the role name and ID, updates, and deletions.
+func (s *HTTPServer) ACLRoleSpecificRequest(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
+
+	// Grab the suffix of the request, so we can further understand it.
+	reqSuffix := strings.TrimPrefix(req.URL.Path, "/v1/acl/role/")
+
+	// Split the request suffix in order to identify whether this is a lookup
+	// of a service, or whether this includes a service and service identifier.
+	suffixParts := strings.Split(reqSuffix, "/")
+
+	switch len(suffixParts) {
+	case 1:
+		// Ensure the role ID is not an empty string which is possible if the
+		// caller requested "/v1/acl/role/"
+		if suffixParts[0] == "" {
+			return nil, CodedError(http.StatusBadRequest, "missing ACL role ID")
+		}
+		return s.aclRoleRequest(resp, req, suffixParts[0])
+	case 2:
+		// This endpoint only supports GET.
+		if req.Method != http.MethodGet {
+			return nil, CodedError(http.StatusMethodNotAllowed, ErrInvalidMethod)
+		}
+
+		// Ensure that the path is correct, otherwise the call could use
+		// "/v1/acl/role/foobar/role-name" and successfully pass through here.
+		if suffixParts[0] != "name" {
+			return nil, CodedError(http.StatusBadRequest, "invalid URI")
+		}
+
+		// Ensure the role name is not an empty string which is possible if the
+		// caller requested "/v1/acl/role/name/"
+		if suffixParts[1] == "" {
+			return nil, CodedError(http.StatusBadRequest, "missing ACL role name")
+		}
+
+		return s.aclRoleGetByNameRequest(resp, req, suffixParts[1])
+
+	default:
+		return nil, CodedError(http.StatusBadRequest, "invalid URI")
+	}
+}
+
+func (s *HTTPServer) aclRoleRequest(
+	resp http.ResponseWriter, req *http.Request, roleID string) (interface{}, error) {
+
+	// Identify the method which indicates which downstream function should be
+	// called.
+	switch req.Method {
+	case http.MethodGet:
+		return s.aclRoleGetByIDRequest(resp, req, roleID)
+	case http.MethodDelete:
+		return s.aclRoleDeleteRequest(resp, req, roleID)
+	case http.MethodPost, http.MethodPut:
+		return s.aclRoleUpsertRequest(resp, req, roleID)
+	default:
+		return nil, CodedError(http.StatusMethodNotAllowed, ErrInvalidMethod)
+	}
+}
+
+func (s *HTTPServer) aclRoleGetByIDRequest(
+	resp http.ResponseWriter, req *http.Request, roleID string) (interface{}, error) {
+
+	args := structs.ACLRoleByIDRequest{
+		RoleID: roleID,
+	}
+	if s.parse(resp, req, &args.Region, &args.QueryOptions) {
+		return nil, nil
+	}
+
+	var reply structs.ACLRoleByIDResponse
+	if err := s.agent.RPC(structs.ACLGetRoleByIDRPCMethod, &args, &reply); err != nil {
+		return nil, err
+	}
+	setMeta(resp, &reply.QueryMeta)
+
+	if reply.ACLRole == nil {
+		return nil, CodedError(http.StatusNotFound, "ACL role not found")
+	}
+	return reply.ACLRole, nil
+}
+
+func (s *HTTPServer) aclRoleDeleteRequest(
+	resp http.ResponseWriter, req *http.Request, roleID string) (interface{}, error) {
+
+	args := structs.ACLRolesDeleteByIDRequest{
+		ACLRoleIDs: []string{roleID},
+	}
+	s.parseWriteRequest(req, &args.WriteRequest)
+
+	var reply structs.ACLRolesDeleteByIDResponse
+	if err := s.agent.RPC(structs.ACLDeleteRolesByIDRPCMethod, &args, &reply); err != nil {
+		return nil, err
+	}
+	setIndex(resp, reply.Index)
+	return nil, nil
+
+}
+
+// aclRoleUpsertRequest handles upserting an ACL to the Nomad servers. It can
+// handle both new creations, and updates to existing roles.
+func (s *HTTPServer) aclRoleUpsertRequest(
+	resp http.ResponseWriter, req *http.Request, roleID string) (interface{}, error) {
+
+	// Decode the ACL role.
+	var aclRole structs.ACLRole
+	if err := decodeBody(req, &aclRole); err != nil {
+		return nil, CodedError(http.StatusInternalServerError, err.Error())
+	}
+
+	// Ensure the request path ID matches the ACL role ID that was decoded.
+	// Only perform this check on updates as a generic error on creation might
+	// be confusing to operators as there is no specific role request path.
+	if roleID != "" && roleID != aclRole.ID {
+		return nil, CodedError(http.StatusBadRequest, "ACL role ID does not match request path")
+	}
+
+	args := structs.ACLRolesUpsertRequest{
+		ACLRoles: []*structs.ACLRole{&aclRole},
+	}
+	s.parseWriteRequest(req, &args.WriteRequest)
+
+	var out structs.ACLRolesUpsertResponse
+	if err := s.agent.RPC(structs.ACLUpsertRolesRPCMethod, &args, &out); err != nil {
+		return nil, err
+	}
+	setIndex(resp, out.Index)
+
+	if len(out.ACLRoles) > 0 {
+		return out.ACLRoles[0], nil
+	}
+	return nil, nil
+
+}
+
+func (s *HTTPServer) aclRoleGetByNameRequest(
+	resp http.ResponseWriter, req *http.Request, roleName string) (interface{}, error) {
+
+	args := structs.ACLRoleByNameRequest{
+		RoleName: roleName,
+	}
+	if s.parse(resp, req, &args.Region, &args.QueryOptions) {
+		return nil, nil
+	}
+
+	var reply structs.ACLRoleByNameResponse
+	if err := s.agent.RPC(structs.ACLGetRoleByNameRPCMethod, &args, &reply); err != nil {
+		return nil, err
+	}
+	setMeta(resp, &reply.QueryMeta)
+
+	if reply.ACLRole == nil {
+		return nil, CodedError(http.StatusNotFound, "ACL role not found")
+	}
+	return reply.ACLRole, nil
+}
diff --git a/command/agent/acl_endpoint_test.go b/command/agent/acl_endpoint_test.go
index 0e52a0899..25350fb0f 100644
--- a/command/agent/acl_endpoint_test.go
+++ b/command/agent/acl_endpoint_test.go
@@ -1,11 +1,13 @@
 package agent
 
 import (
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 
 	"github.com/hashicorp/nomad/ci"
+	"github.com/hashicorp/nomad/helper/uuid"
 	"github.com/hashicorp/nomad/nomad/mock"
 	"github.com/hashicorp/nomad/nomad/structs"
 	"github.com/stretchr/testify/assert"
@@ -558,3 +560,438 @@ func TestHTTP_OneTimeToken(t *testing.T) {
 		require.EqualError(t, err, structs.ErrPermissionDenied.Error())
 	})
 }
+
+func TestHTTPServer_ACLRoleListRequest(t *testing.T) {
+	ci.Parallel(t)
+
+	testCases := []struct {
+		name   string
+		testFn func(srv *TestAgent)
+	}{
+		{
+			name: "no auth token set",
+			testFn: func(srv *TestAgent) {
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodGet, "/v1/acl/roles", nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleListRequest(respW, req)
+				require.Error(t, err)
+				require.ErrorContains(t, err, "Permission denied")
+				require.Nil(t, obj)
+			},
+		},
+		{
+			name: "invalid method",
+			testFn: func(srv *TestAgent) {
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodConnect, "/v1/acl/roles", nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Ensure we have a token set.
+				setToken(req, srv.RootToken)
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleListRequest(respW, req)
+				require.Error(t, err)
+				require.ErrorContains(t, err, "Invalid method")
+				require.Nil(t, obj)
+			},
+		},
+		{
+			name: "no roles in state",
+			testFn: func(srv *TestAgent) {
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodGet, "/v1/acl/roles", nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Ensure we have a token set.
+				setToken(req, srv.RootToken)
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleListRequest(respW, req)
+				require.NoError(t, err)
+				require.Empty(t, obj.([]*structs.ACLRole))
+			},
+		},
+		{
+			name: "roles in state",
+			testFn: func(srv *TestAgent) {
+
+				// 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, srv.server.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, srv.server.State().UpsertACLRoles(structs.MsgTypeTestSetup, 20, aclRoles))
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodGet, "/v1/acl/roles", nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Ensure we have a token set.
+				setToken(req, srv.RootToken)
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleListRequest(respW, req)
+				require.NoError(t, err)
+				require.Len(t, obj.([]*structs.ACLRole), 2)
+			},
+		},
+		{
+			name: "roles in state using prefix",
+			testFn: func(srv *TestAgent) {
+
+				// 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, srv.server.State().UpsertACLPolicies(
+					structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+				// Create two ACL roles and put these directly into state, one
+				// using a custom prefix.
+				aclRoles := []*structs.ACLRole{mock.ACLRole(), mock.ACLRole()}
+				aclRoles[1].ID = "badger-badger-badger-" + uuid.Generate()
+				require.NoError(t, srv.server.State().UpsertACLRoles(structs.MsgTypeTestSetup, 20, aclRoles))
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodGet, "/v1/acl/roles?prefix=badger-badger-badger", nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Ensure we have a token set.
+				setToken(req, srv.RootToken)
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleListRequest(respW, req)
+				require.NoError(t, err)
+				require.Len(t, obj.([]*structs.ACLRole), 1)
+				require.Contains(t, obj.([]*structs.ACLRole)[0].ID, "badger-badger-badger")
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			httpACLTest(t, nil, tc.testFn)
+		})
+	}
+}
+
+func TestHTTPServer_ACLRoleRequest(t *testing.T) {
+	ci.Parallel(t)
+
+	testCases := []struct {
+		name   string
+		testFn func(srv *TestAgent)
+	}{
+		{
+			name: "no auth token set",
+			testFn: func(srv *TestAgent) {
+
+				// Create a mock role to use in the request body.
+				mockACLRole := mock.ACLRole()
+				mockACLRole.ID = ""
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodPut, "/v1/acl/role", encodeReq(mockACLRole))
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleRequest(respW, req)
+				require.Error(t, err)
+				require.ErrorContains(t, err, "Permission denied")
+				require.Nil(t, obj)
+			},
+		},
+		{
+			name: "invalid method",
+			testFn: func(srv *TestAgent) {
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodConnect, "/v1/acl/role", nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Ensure we have a token set.
+				setToken(req, srv.RootToken)
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleRequest(respW, req)
+				require.Error(t, err)
+				require.ErrorContains(t, err, "Invalid method")
+				require.Nil(t, obj)
+			},
+		},
+		{
+			name: "successful upsert",
+			testFn: func(srv *TestAgent) {
+
+				// 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, srv.server.State().UpsertACLPolicies(
+					structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+				// Create a mock role to use in the request body.
+				mockACLRole := mock.ACLRole()
+				mockACLRole.ID = ""
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodPut, "/v1/acl/role", encodeReq(mockACLRole))
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Ensure we have a token set.
+				setToken(req, srv.RootToken)
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleRequest(respW, req)
+				require.NoError(t, err)
+				require.NotNil(t, obj)
+				require.Equal(t, obj.(*structs.ACLRole).Hash, mockACLRole.Hash)
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			httpACLTest(t, nil, tc.testFn)
+		})
+	}
+}
+
+func TestHTTPServer_ACLRoleSpecificRequest(t *testing.T) {
+	ci.Parallel(t)
+
+	testCases := []struct {
+		name   string
+		testFn func(srv *TestAgent)
+	}{
+		{
+			name: "invalid URI",
+			testFn: func(srv *TestAgent) {
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodGet, "/v1/acl/role/name/this/is/will/not/work", nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleSpecificRequest(respW, req)
+				require.Error(t, err)
+				require.ErrorContains(t, err, "invalid URI")
+				require.Nil(t, obj)
+			},
+		},
+		{
+			name: "invalid role name lookalike URI",
+			testFn: func(srv *TestAgent) {
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodGet, "/v1/acl/role/foobar/rolename", nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleSpecificRequest(respW, req)
+				require.Error(t, err)
+				require.ErrorContains(t, err, "invalid URI")
+				require.Nil(t, obj)
+			},
+		},
+		{
+			name: "missing role name",
+			testFn: func(srv *TestAgent) {
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodGet, "/v1/acl/role/name/", nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleSpecificRequest(respW, req)
+				require.Error(t, err)
+				require.ErrorContains(t, err, "missing ACL role name")
+				require.Nil(t, obj)
+			},
+		},
+		{
+			name: "missing role ID",
+			testFn: func(srv *TestAgent) {
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodGet, "/v1/acl/role/", nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleSpecificRequest(respW, req)
+				require.Error(t, err)
+				require.ErrorContains(t, err, "missing ACL role ID")
+				require.Nil(t, obj)
+			},
+		},
+		{
+			name: "role name incorrect method",
+			testFn: func(srv *TestAgent) {
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodConnect, "/v1/acl/role/name/foobar", nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleSpecificRequest(respW, req)
+				require.Error(t, err)
+				require.ErrorContains(t, err, "Invalid method")
+				require.Nil(t, obj)
+			},
+		},
+		{
+			name: "role ID incorrect method",
+			testFn: func(srv *TestAgent) {
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodConnect, "/v1/acl/role/foobar", nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleSpecificRequest(respW, req)
+				require.Error(t, err)
+				require.ErrorContains(t, err, "Invalid method")
+				require.Nil(t, obj)
+			},
+		},
+		{
+			name: "get role by name",
+			testFn: func(srv *TestAgent) {
+
+				// 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, srv.server.State().UpsertACLPolicies(
+					structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+				// Create a mock role and put directly into state.
+				mockACLRole := mock.ACLRole()
+				require.NoError(t, srv.server.State().UpsertACLRoles(
+					structs.MsgTypeTestSetup, 20, []*structs.ACLRole{mockACLRole}))
+
+				url := fmt.Sprintf("/v1/acl/role/name/%s", mockACLRole.Name)
+
+				// Build the HTTP request.
+				req, err := http.NewRequest(http.MethodGet, url, nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Ensure we have a token set.
+				setToken(req, srv.RootToken)
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleSpecificRequest(respW, req)
+				require.NoError(t, err)
+				require.Equal(t, obj.(*structs.ACLRole).Hash, mockACLRole.Hash)
+			},
+		},
+		{
+			name: "get, update, and delete role by ID",
+			testFn: func(srv *TestAgent) {
+
+				// 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, srv.server.State().UpsertACLPolicies(
+					structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+				// Create a mock role and put directly into state.
+				mockACLRole := mock.ACLRole()
+				require.NoError(t, srv.server.State().UpsertACLRoles(
+					structs.MsgTypeTestSetup, 20, []*structs.ACLRole{mockACLRole}))
+
+				url := fmt.Sprintf("/v1/acl/role/%s", mockACLRole.ID)
+
+				// Build the HTTP request to read the role using its ID.
+				req, err := http.NewRequest(http.MethodGet, url, nil)
+				require.NoError(t, err)
+				respW := httptest.NewRecorder()
+
+				// Ensure we have a token set.
+				setToken(req, srv.RootToken)
+
+				// Send the HTTP request.
+				obj, err := srv.Server.ACLRoleSpecificRequest(respW, req)
+				require.NoError(t, err)
+				require.Equal(t, obj.(*structs.ACLRole).Hash, mockACLRole.Hash)
+
+				// Update the role policy list and make the request via the
+				// HTTP API.
+				mockACLRole.Policies = []*structs.ACLRolePolicyLink{{Name: "mocked-test-policy-1"}}
+
+				req, err = http.NewRequest(http.MethodPost, url, encodeReq(mockACLRole))
+				require.NoError(t, err)
+				respW = httptest.NewRecorder()
+
+				// Ensure we have a token set.
+				setToken(req, srv.RootToken)
+
+				// Send the HTTP request.
+				obj, err = srv.Server.ACLRoleSpecificRequest(respW, req)
+				require.NoError(t, err)
+				require.Equal(t, obj.(*structs.ACLRole).Policies, mockACLRole.Policies)
+
+				// Delete the ACL role using its ID.
+				req, err = http.NewRequest(http.MethodDelete, url, nil)
+				require.NoError(t, err)
+				respW = httptest.NewRecorder()
+
+				// Ensure we have a token set.
+				setToken(req, srv.RootToken)
+
+				// Send the HTTP request.
+				obj, err = srv.Server.ACLRoleSpecificRequest(respW, req)
+				require.NoError(t, err)
+				require.Nil(t, obj)
+
+				// Ensure the ACL role is no longer stored within state.
+				aclRole, err := srv.server.State().GetACLRoleByID(nil, mockACLRole.ID)
+				require.NoError(t, err)
+				require.Nil(t, aclRole)
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			httpACLTest(t, nil, tc.testFn)
+		})
+	}
+}
diff --git a/command/agent/http.go b/command/agent/http.go
index b0676ec82..8a73e786a 100644
--- a/command/agent/http.go
+++ b/command/agent/http.go
@@ -381,6 +381,11 @@ func (s HTTPServer) registerHandlers(enableDebug bool) {
 	s.mux.HandleFunc("/v1/acl/token", s.wrap(s.ACLTokenSpecificRequest))
 	s.mux.HandleFunc("/v1/acl/token/", s.wrap(s.ACLTokenSpecificRequest))
 
+	// Register our ACL role handlers.
+	s.mux.HandleFunc("/v1/acl/roles", s.wrap(s.ACLRoleListRequest))
+	s.mux.HandleFunc("/v1/acl/role", s.wrap(s.ACLRoleRequest))
+	s.mux.HandleFunc("/v1/acl/role/", s.wrap(s.ACLRoleSpecificRequest))
+
 	s.mux.Handle("/v1/client/fs/", wrapCORS(s.wrap(s.FsRequest)))
 	s.mux.HandleFunc("/v1/client/gc", s.wrap(s.ClientGCRequest))
 	s.mux.Handle("/v1/client/stats", wrapCORS(s.wrap(s.ClientStatsRequest)))
diff --git a/nomad/acl_endpoint.go b/nomad/acl_endpoint.go
index f4958df71..70b386f49 100644
--- 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
+		},
+	})
+}
diff --git a/nomad/acl_endpoint_test.go b/nomad/acl_endpoint_test.go
index 87884d9bf..a3ff68460 100644
--- 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]))
+}
diff --git a/nomad/state/state_store_acl.go b/nomad/state/state_store_acl.go
index 9fd579cf6..60df4e506 100644
--- 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
+}
diff --git a/nomad/state/state_store_acl_test.go b/nomad/state/state_store_acl_test.go
index 007de2b04..6cc6a0ae4 100644
--- 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)
+}
diff --git a/nomad/structs/acl.go b/nomad/structs/acl.go
index 6ce3c5b0c..e40a89d42 100644
--- 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
+}
diff --git a/nomad/structs/acl_test.go b/nomad/structs/acl_test.go
index d2621d415..22e263e5e 100644
--- 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())
+}