diff --git a/command/agent/consul/acl_testing.go b/command/agent/consul/acl_testing.go
index 8bd0821d6ec..498cfe038fd 100644
--- a/command/agent/consul/acl_testing.go
+++ b/command/agent/consul/acl_testing.go
@@ -119,8 +119,8 @@ func (m *MockACLsAPI) RoleRead(roleID string, _ *api.QueryOptions) (*api.ACLRole
 	}
 }
 
-// Example Consul "operator" tokens for use in tests.
-
+// Example Consul ACL tokens for use in tests. These tokens belong to the
+// default Consul namespace.
 const (
 	ExampleOperatorTokenID0 = "de591604-86eb-1e6f-8b44-d4db752921ae"
 	ExampleOperatorTokenID1 = "59c219c2-47e4-43f3-bb45-258fd13f59d5"
@@ -130,7 +130,20 @@ const (
 	ExampleOperatorTokenID5 = "097cbb45-506b-c79c-ec38-82eb0dc0794a"
 )
 
+// Example Consul ACL tokens for use in tests that match the policies as the
+// tokens above, but these belong to the "banana' Consul namespace.
+const (
+	ExampleOperatorTokenID10 = "ddfe688f-655f-e8dd-1db5-5650eed00aeb"
+	ExampleOperatorTokenID11 = "46d09394-598c-1e55-b7fd-64cd2f409707"
+	ExampleOperatorTokenID12 = "a041cb88-0f4b-0314-89f6-10e1e093d2e5"
+	ExampleOperatorTokenID13 = "cc22a583-243f-3258-14ad-db0e56749657"
+	ExampleOperatorTokenID14 = "5b6d0508-13a6-4bc3-33a1-ba1941e1175b"
+	ExampleOperatorTokenID15 = "e9db1754-c075-d0fc-0a7e-de1e9e7bff98"
+)
+
 var (
+	// In Consul namespace "default"
+
 	ExampleOperatorToken0 = &api.ACLToken{
 		SecretID:    ExampleOperatorTokenID0,
 		AccessorID:  "228865c6-3bf6-6683-df03-06dea2779088 ",
@@ -189,6 +202,67 @@ var (
 		}},
 		Namespace: "default",
 	}
+
+	// In Consul namespace "banana"
+
+	ExampleOperatorToken10 = &api.ACLToken{
+		SecretID:    ExampleOperatorTokenID0,
+		AccessorID:  "76a2c3b5-5d64-9089-f701-660eec2d3554",
+		Description: "Operator Token 0",
+		Namespace:   "banana",
+	}
+
+	ExampleOperatorToken11 = &api.ACLToken{
+		SecretID:    ExampleOperatorTokenID1,
+		AccessorID:  "40f2a36a-0a65-1972-106c-b2e5dd46d6e8",
+		Description: "Operator Token 1",
+		Policies: []*api.ACLTokenPolicyLink{{
+			ID: ExamplePolicyID1,
+		}},
+		Namespace: "banana",
+	}
+
+	ExampleOperatorToken12 = &api.ACLToken{
+		SecretID:    ExampleOperatorTokenID2,
+		AccessorID:  "894f2c5c-b285-71bf-4acb-6344cecf71f3",
+		Description: "Operator Token 2",
+		Policies: []*api.ACLTokenPolicyLink{{
+			ID: ExamplePolicyID2,
+		}},
+		Namespace: "banana",
+	}
+
+	ExampleOperatorToken13 = &api.ACLToken{
+		SecretID:    ExampleOperatorTokenID3,
+		AccessorID:  "2a81ec0b-692e-845e-f5b8-c33c05e5af22",
+		Description: "Operator Token 3",
+		Policies: []*api.ACLTokenPolicyLink{{
+			ID: ExamplePolicyID3,
+		}},
+		Namespace: "banana",
+	}
+
+	ExampleOperatorToken14 = &api.ACLToken{
+		SecretID:    ExampleOperatorTokenID4,
+		AccessorID:  "4273f1cc-5626-7a77-dc65-1f24af035ed5d",
+		Description: "Operator Token 4",
+		Policies:    nil, // no direct policy, only roles
+		Roles: []*api.ACLTokenRoleLink{{
+			ID:   ExampleRoleID1,
+			Name: "example-role-1",
+		}},
+		Namespace: "banana",
+	}
+
+	ExampleOperatorToken15 = &api.ACLToken{
+		SecretID:    ExampleOperatorTokenID5,
+		AccessorID:  "5b78e186-87d8-c1ad-966f-f5fa87b05c9a",
+		Description: "Operator Token 5",
+		Policies: []*api.ACLTokenPolicyLink{{
+			ID: ExamplePolicyID4,
+		}},
+		Namespace: "banana",
+	}
 )
 
 func (m *MockACLsAPI) TokenReadSelf(q *api.QueryOptions) (*api.ACLToken, *api.QueryMeta, error) {
@@ -209,6 +283,21 @@ func (m *MockACLsAPI) TokenReadSelf(q *api.QueryOptions) (*api.ACLToken, *api.Qu
 	case ExampleOperatorTokenID5:
 		return ExampleOperatorToken5, nil, nil
 
+	case ExampleOperatorTokenID11:
+		return ExampleOperatorToken11, nil, nil
+
+	case ExampleOperatorTokenID12:
+		return ExampleOperatorToken12, nil, nil
+
+	case ExampleOperatorTokenID13:
+		return ExampleOperatorToken13, nil, nil
+
+	case ExampleOperatorTokenID14:
+		return ExampleOperatorToken14, nil, nil
+
+	case ExampleOperatorTokenID15:
+		return ExampleOperatorToken15, nil, nil
+
 	default:
 		return nil, nil, errors.New("no such token")
 	}
diff --git a/e2e/connect/acls.go b/e2e/connect/acls.go
index 51c4dbfd0b6..2a455c9373a 100644
--- a/e2e/connect/acls.go
+++ b/e2e/connect/acls.go
@@ -22,9 +22,10 @@ type ConnectACLsE2ETest struct {
 	// manageConsulACLs is used to 'enable' and 'disable' Consul ACLs in the
 	// Consul Cluster that has been setup for e2e testing.
 	manageConsulACLs consulacls.Manager
-	// consulMasterToken is set to the generated Consul ACL token after using
+
+	// consulManagementToken is set to the generated Consul ACL token after using
 	// the consul-acls-manage.sh script to enable ACLs.
-	consulMasterToken string
+	consulManagementToken string
 
 	// things to cleanup after each test case
 	jobIDs          []string
@@ -46,7 +47,7 @@ func (tc *ConnectACLsE2ETest) BeforeAll(f *framework.F) {
 
 	// Validate the consul master token exists, otherwise tests are just
 	// going to be a train wreck.
-	tokenLength := len(tc.consulMasterToken)
+	tokenLength := len(tc.consulManagementToken)
 	require.Equal(f.T(), 36, tokenLength, "consul master token wrong length")
 
 	// Validate the CONSUL_HTTP_TOKEN is NOT set, because that will cause
@@ -63,7 +64,7 @@ func (tc *ConnectACLsE2ETest) BeforeAll(f *framework.F) {
 // enableConsulACLs effectively executes `consul-acls-manage.sh enable`, which
 // will activate Consul ACLs, going through the bootstrap process if necessary.
 func (tc *ConnectACLsE2ETest) enableConsulACLs(f *framework.F) {
-	tc.consulMasterToken = tc.manageConsulACLs.Enable(f.T())
+	tc.consulManagementToken = tc.manageConsulACLs.Enable(f.T())
 }
 
 // AfterAll runs after all tests are complete.
@@ -100,14 +101,14 @@ func (tc *ConnectACLsE2ETest) AfterEach(f *framework.F) {
 	// cleanup consul tokens
 	for _, id := range tc.consulTokenIDs {
 		t.Log("cleanup: delete consul token id:", id)
-		_, err := tc.Consul().ACL().TokenDelete(id, &consulapi.WriteOptions{Token: tc.consulMasterToken})
+		_, err := tc.Consul().ACL().TokenDelete(id, &consulapi.WriteOptions{Token: tc.consulManagementToken})
 		f.NoError(err)
 	}
 
 	// cleanup consul policies
 	for _, id := range tc.consulPolicyIDs {
 		t.Log("cleanup: delete consul policy id:", id)
-		_, err := tc.Consul().ACL().PolicyDelete(id, &consulapi.WriteOptions{Token: tc.consulMasterToken})
+		_, err := tc.Consul().ACL().PolicyDelete(id, &consulapi.WriteOptions{Token: tc.consulManagementToken})
 		f.NoError(err)
 	}
 
@@ -128,27 +129,30 @@ func (tc *ConnectACLsE2ETest) AfterEach(f *framework.F) {
 	tc.consulPolicyIDs = []string{}
 }
 
+// todo(shoenig): follow up refactor with e2eutil.ConsulPolicy
 type consulPolicy struct {
 	Name  string // e.g. nomad-operator
 	Rules string // e.g. service "" { policy="write" }
 }
 
+// todo(shoenig): follow up refactor with e2eutil.ConsulPolicy
 func (tc *ConnectACLsE2ETest) createConsulPolicy(p consulPolicy, f *framework.F) string {
 	result, _, err := tc.Consul().ACL().PolicyCreate(&consulapi.ACLPolicy{
 		Name:        p.Name,
 		Description: "test policy " + p.Name,
 		Rules:       p.Rules,
-	}, &consulapi.WriteOptions{Token: tc.consulMasterToken})
+	}, &consulapi.WriteOptions{Token: tc.consulManagementToken})
 	f.NoError(err, "failed to create consul policy")
 	tc.consulPolicyIDs = append(tc.consulPolicyIDs, result.ID)
 	return result.ID
 }
 
+// todo(shoenig): follow up refactor with e2eutil.ConsulPolicy
 func (tc *ConnectACLsE2ETest) createOperatorToken(policyID string, f *framework.F) string {
 	token, _, err := tc.Consul().ACL().TokenCreate(&consulapi.ACLToken{
 		Description: "operator token",
 		Policies:    []*consulapi.ACLTokenPolicyLink{{ID: policyID}},
-	}, &consulapi.WriteOptions{Token: tc.consulMasterToken})
+	}, &consulapi.WriteOptions{Token: tc.consulManagementToken})
 	f.NoError(err, "failed to create operator token")
 	tc.consulTokenIDs = append(tc.consulTokenIDs, token.AccessorID)
 	return token.SecretID
@@ -170,7 +174,7 @@ func (tc *ConnectACLsE2ETest) TestConnectACLsRegisterMasterToken(f *framework.F)
 	// Set the job file to use the consul master token.
 	// One should never do this in practice, but, it should work.
 	// https://www.consul.io/docs/acl/acl-system.html#builtin-tokens
-	job.ConsulToken = &tc.consulMasterToken
+	job.ConsulToken = &tc.consulManagementToken
 
 	// Avoid using Register here, because that would actually create and run the
 	// Job which runs the task, creates the SI token, which all needs to be
@@ -368,7 +372,7 @@ func (tc *ConnectACLsE2ETest) serviceofSIToken(description string) string {
 func (tc *ConnectACLsE2ETest) countSITokens(t *testing.T) map[string]int {
 	aclAPI := tc.Consul().ACL()
 	tokens, _, err := aclAPI.TokenList(&consulapi.QueryOptions{
-		Token: tc.consulMasterToken,
+		Token: tc.consulManagementToken,
 	})
 	require.NoError(t, err)
 
diff --git a/e2e/connect/connect.go b/e2e/connect/connect.go
index e6e578ceba3..d224b01613d 100644
--- a/e2e/connect/connect.go
+++ b/e2e/connect/connect.go
@@ -50,11 +50,11 @@ func init() {
 	})
 
 	// Connect tests with Consul ACLs enabled. These are now gated behind the
-	// NOMAD_TEST_CONNECT_ACLS environment variable, because they cause lots of
+	// NOMAD_TEST_CONSUL_ACLS environment variable, because they cause lots of
 	// problems for e2e test flakiness (due to restarting consul, nomad, etc.).
 	//
 	// Run these tests locally when working on Connect.
-	if os.Getenv("NOMAD_TEST_CONNECT_ACLS") == "1" {
+	if os.Getenv("NOMAD_TEST_CONSUL_ACLS") == "1" {
 		framework.AddSuites(&framework.TestSuite{
 			Component:   "ConnectACLs",
 			CanRunLocal: false,
diff --git a/e2e/consul/namespaces.go b/e2e/consul/namespaces.go
index 74f89adb290..407e6dfb031 100644
--- a/e2e/consul/namespaces.go
+++ b/e2e/consul/namespaces.go
@@ -39,7 +39,12 @@ var (
 
 type ConsulNamespacesE2ETest struct {
 	framework.TC
+
 	jobIDs []string
+
+	// cToken contains the Consul global-management token during ACL enabled
+	// tests (i.e. ConsulNamespacesE2ETestACLs which embeds ConsulNamespacesE2ETest).
+	cToken string
 }
 
 func (tc *ConsulNamespacesE2ETest) BeforeAll(f *framework.F) {
@@ -56,6 +61,9 @@ func (tc *ConsulNamespacesE2ETest) BeforeAll(f *framework.F) {
 		value := fmt.Sprintf("ns_%s", namespace)
 		e2eutil.PutConsulKey(f.T(), tc.Consul(), namespace, "ns-kv-example", value)
 	}
+
+	// make the unused variable linter happy in oss
+	f.T().Log("Consul global-management token:", tc.cToken)
 }
 
 func (tc *ConsulNamespacesE2ETest) AfterAll(f *framework.F) {
@@ -68,13 +76,13 @@ func (tc *ConsulNamespacesE2ETest) TestNamespacesExist(f *framework.F) {
 	require.True(f.T(), helper.CompareSliceSetString(namespaces, append(consulNamespaces, "default")))
 }
 
-func (tc *ConsulNamespacesE2ETest) testConsulRegisterGroupServices(f *framework.F, nsA, nsB, nsC, nsZ string) {
+func (tc *ConsulNamespacesE2ETest) testConsulRegisterGroupServices(f *framework.F, token, nsA, nsB, nsC, nsZ string) {
 	nomadClient := tc.Nomad()
 	jobID := "cns-group-services"
 	tc.jobIDs = append(tc.jobIDs, jobID)
 
 	// Run job and wait for allocs
-	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobGroupServices, jobID, "")
+	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobGroupServices, jobID, token)
 	require.Len(f.T(), allocations, 3)
 	allocIDs := e2eutil.AllocIDsFromAllocationListStubs(allocations)
 	e2eutil.WaitForAllocsRunning(f.T(), tc.Nomad(), allocIDs)
@@ -112,13 +120,13 @@ func (tc *ConsulNamespacesE2ETest) testConsulRegisterGroupServices(f *framework.
 	e2eutil.RequireConsulDeregistered(r, c, nsZ, "z2")
 }
 
-func (tc *ConsulNamespacesE2ETest) testConsulRegisterTaskServices(f *framework.F, nsA, nsB, nsC, nsZ string) {
+func (tc *ConsulNamespacesE2ETest) testConsulRegisterTaskServices(f *framework.F, token, nsA, nsB, nsC, nsZ string) {
 	nomadClient := tc.Nomad()
 	jobID := "cns-task-services"
 	tc.jobIDs = append(tc.jobIDs, jobID)
 
 	// Run job and wait for allocs
-	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobTaskServices, jobID, "")
+	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobTaskServices, jobID, token)
 	require.Len(f.T(), allocations, 3)
 	allocIDs := e2eutil.AllocIDsFromAllocationListStubs(allocations)
 	e2eutil.WaitForAllocsRunning(f.T(), tc.Nomad(), allocIDs)
@@ -154,14 +162,14 @@ func (tc *ConsulNamespacesE2ETest) testConsulRegisterTaskServices(f *framework.F
 	e2eutil.RequireConsulDeregistered(r, c, nsZ, "z2")
 }
 
-func (tc *ConsulNamespacesE2ETest) testConsulTemplateKV(f *framework.F, expB, expZ string) {
+func (tc *ConsulNamespacesE2ETest) testConsulTemplateKV(f *framework.F, token, expB, expZ string) {
 	t := f.T()
 	nomadClient := tc.Nomad()
 	jobID := "cns-template-kv"
 	tc.jobIDs = append(tc.jobIDs, jobID)
 
 	// Run job and wait for allocs to complete
-	allocations := e2eutil.RegisterAndWaitForAllocs(t, nomadClient, cnsJobTemplateKV, jobID, "")
+	allocations := e2eutil.RegisterAndWaitForAllocs(t, nomadClient, cnsJobTemplateKV, jobID, token)
 	require.Len(t, allocations, 2)
 	allocIDs := e2eutil.AllocIDsFromAllocationListStubs(allocations)
 	e2eutil.WaitForAllocsStopped(f.T(), tc.Nomad(), allocIDs)
@@ -183,13 +191,13 @@ func (tc *ConsulNamespacesE2ETest) testConsulTemplateKV(f *framework.F, expB, ex
 	e2eutil.WaitForJobStopped(t, nomadClient, jobID)
 }
 
-func (tc *ConsulNamespacesE2ETest) testConsulConnectSidecars(f *framework.F, nsA, nsZ string) {
+func (tc *ConsulNamespacesE2ETest) testConsulConnectSidecars(f *framework.F, token, nsA, nsZ string) {
 	nomadClient := tc.Nomad()
 	jobID := "cns-connect-sidecars"
 	tc.jobIDs = append(tc.jobIDs, jobID)
 
 	// Run job and wait for allocs
-	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobConnectSidecars, jobID, "")
+	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobConnectSidecars, jobID, token)
 	require.Len(f.T(), allocations, 4)
 	allocIDs := e2eutil.AllocIDsFromAllocationListStubs(allocations)
 	e2eutil.WaitForAllocsRunning(f.T(), tc.Nomad(), allocIDs)
@@ -223,13 +231,13 @@ func (tc *ConsulNamespacesE2ETest) testConsulConnectSidecars(f *framework.F, nsA
 	e2eutil.RequireConsulDeregistered(r, c, nsZ, "count-dashboard-z-sidecar-proxy")
 }
 
-func (tc *ConsulNamespacesE2ETest) testConsulConnectIngressGateway(f *framework.F, nsA, nsZ string) {
+func (tc *ConsulNamespacesE2ETest) testConsulConnectIngressGateway(f *framework.F, token, nsA, nsZ string) {
 	nomadClient := tc.Nomad()
 	jobID := "cns-connect-ingress"
 	tc.jobIDs = append(tc.jobIDs, jobID)
 
 	// Run job and wait for allocs
-	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobConnectIngress, jobID, "")
+	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobConnectIngress, jobID, token)
 	require.Len(f.T(), allocations, 4) // 2 x (1 service + 1 gateway)
 	allocIDs := e2eutil.AllocIDsFromAllocationListStubs(allocations)
 	e2eutil.WaitForAllocsRunning(f.T(), tc.Nomad(), allocIDs)
@@ -261,13 +269,13 @@ func (tc *ConsulNamespacesE2ETest) testConsulConnectIngressGateway(f *framework.
 	e2eutil.DeleteConsulConfigEntry(f.T(), c, nsZ, "ingress-gateway", "my-ingress-service-z")
 }
 
-func (tc *ConsulNamespacesE2ETest) testConsulConnectTerminatingGateway(f *framework.F, nsA, nsZ string) {
+func (tc *ConsulNamespacesE2ETest) testConsulConnectTerminatingGateway(f *framework.F, token, nsA, nsZ string) {
 	nomadClient := tc.Nomad()
 	jobID := "cns-connect-terminating"
 	tc.jobIDs = append(tc.jobIDs, jobID)
 
 	// Run job and wait for allocs
-	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobConnectTerminating, jobID, "")
+	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobConnectTerminating, jobID, token)
 	require.Len(f.T(), allocations, 6) // 2 x (2 services + 1 gateway)
 	allocIDs := e2eutil.AllocIDsFromAllocationListStubs(allocations)
 	e2eutil.WaitForAllocsRunning(f.T(), tc.Nomad(), allocIDs)
@@ -301,13 +309,13 @@ func (tc *ConsulNamespacesE2ETest) testConsulConnectTerminatingGateway(f *framew
 	e2eutil.DeleteConsulConfigEntry(f.T(), c, nsZ, "terminating-gateway", "api-gateway-z")
 }
 
-func (tc *ConsulNamespacesE2ETest) testConsulScriptChecksTask(f *framework.F, nsA, nsZ string) {
+func (tc *ConsulNamespacesE2ETest) testConsulScriptChecksTask(f *framework.F, token, nsA, nsZ string) {
 	nomadClient := tc.Nomad()
 	jobID := "cns-script-checks-task"
 	tc.jobIDs = append(tc.jobIDs, jobID)
 
 	// Run job and wait for allocs
-	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobScriptChecksTask, jobID, "")
+	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobScriptChecksTask, jobID, token)
 	require.Len(f.T(), allocations, 2)
 	allocIDs := e2eutil.AllocIDsFromAllocationListStubs(allocations)
 	e2eutil.WaitForAllocsRunning(f.T(), tc.Nomad(), allocIDs)
@@ -351,13 +359,13 @@ func (tc *ConsulNamespacesE2ETest) testConsulScriptChecksTask(f *framework.F, ns
 	e2eutil.WaitForJobStopped(f.T(), nomadClient, jobID)
 }
 
-func (tc *ConsulNamespacesE2ETest) testConsulScriptChecksGroup(f *framework.F, nsA, nsZ string) {
+func (tc *ConsulNamespacesE2ETest) testConsulScriptChecksGroup(f *framework.F, token, nsA, nsZ string) {
 	nomadClient := tc.Nomad()
 	jobID := "cns-script-checks-group"
 	tc.jobIDs = append(tc.jobIDs, jobID)
 
 	// Run job and wait for allocs
-	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobScriptChecksGroup, jobID, "")
+	allocations := e2eutil.RegisterAndWaitForAllocs(f.T(), nomadClient, cnsJobScriptChecksGroup, jobID, token)
 	require.Len(f.T(), allocations, 2)
 	allocIDs := e2eutil.AllocIDsFromAllocationListStubs(allocations)
 	e2eutil.WaitForAllocsRunning(f.T(), tc.Nomad(), allocIDs)
diff --git a/e2e/consulacls/nomad-client-policy.hcl b/e2e/consulacls/nomad-client-policy.hcl
index 3cd9087693e..cd3c3b1c8cd 100644
--- a/e2e/consulacls/nomad-client-policy.hcl
+++ b/e2e/consulacls/nomad-client-policy.hcl
@@ -1,13 +1,22 @@
 // The Nomad Client will be registering things into its buddy Consul Client.
-
-service_prefix "" {
-  policy = "write"
-}
+// Note: because we also test the use of Consul namespaces, this token must be
+// able to register services, read the keystore, and read node data for any
+// namespace.
 
 agent_prefix "" {
   policy = "read"
 }
 
-node_prefix "" {
-  policy = "read"
-}
+namespace_prefix "" {
+  key_prefix "" {
+    policy = "read"
+  }
+
+  node_prefix "" {
+    policy = "read"
+  }
+
+  service_prefix "" {
+    policy = "write"
+  }
+}
\ No newline at end of file
diff --git a/e2e/consulacls/nomad-server-policy.hcl b/e2e/consulacls/nomad-server-policy.hcl
index 4753838a24a..6dd5a64c326 100644
--- a/e2e/consulacls/nomad-server-policy.hcl
+++ b/e2e/consulacls/nomad-server-policy.hcl
@@ -1,11 +1,15 @@
-// The acl=write permission is required for generating Consul Service Identity
-// tokens for consul connect services.
-acl = "write"
-
 // The operator=write permission is required for creating config entries for
-// connect ingress gateways.
+// connect ingress gateways. operator ACLs are not namespaced, though the
+// config entries they can generate are.
 operator = "write"
 
+namespace_prefix "" {
+  // The acl=write permission is required for generating Consul Service Identity
+  // tokens for consul connect services. Those services could be configured for
+  // any Consul namespace the job-submitter has access to.
+  acl = "write"
+}
+
 service_prefix "" {
   policy = "write"
 }
diff --git a/e2e/e2eutil/consul.go b/e2e/e2eutil/consul.go
index d804feee622..27f8cf5abb5 100644
--- a/e2e/e2eutil/consul.go
+++ b/e2e/e2eutil/consul.go
@@ -52,8 +52,8 @@ func RequireConsulDeregistered(require *require.Assertions, client *capi.Client,
 
 // RequireConsulRegistered assert that the service is registered in Consul.
 func RequireConsulRegistered(require *require.Assertions, client *capi.Client, namespace, service string, count int) {
-	testutil.WaitForResultRetries(5, func() (bool, error) {
-		defer time.Sleep(time.Second)
+	testutil.WaitForResultRetries(10, func() (bool, error) {
+		defer time.Sleep(2 * time.Second)
 
 		services, _, err := client.Catalog().Service(service, "", &capi.QueryOptions{Namespace: namespace})
 		require.NoError(err)
@@ -72,6 +72,7 @@ func RequireConsulRegistered(require *require.Assertions, client *capi.Client, n
 // Requires Consul Enterprise.
 func CreateConsulNamespaces(t *testing.T, client *capi.Client, namespaces []string) {
 	nsClient := client.Namespaces()
+
 	for _, namespace := range namespaces {
 		_, _, err := nsClient.Create(&capi.Namespace{
 			Name:        namespace,
@@ -86,6 +87,7 @@ func CreateConsulNamespaces(t *testing.T, client *capi.Client, namespaces []stri
 // Requires Consul Enterprise.
 func DeleteConsulNamespaces(t *testing.T, client *capi.Client, namespaces []string) {
 	nsClient := client.Namespaces()
+
 	for _, namespace := range namespaces {
 		_, err := nsClient.Delete(namespace, nil)
 		assert.NoError(t, err) // be lenient; used in cleanup
@@ -97,8 +99,10 @@ func DeleteConsulNamespaces(t *testing.T, client *capi.Client, namespaces []stri
 // Requires Consul Enterprise.
 func ListConsulNamespaces(t *testing.T, client *capi.Client) []string {
 	nsClient := client.Namespaces()
+
 	namespaces, _, err := nsClient.List(nil)
 	require.NoError(t, err)
+
 	result := make([]string, 0, len(namespaces))
 	for _, namespace := range namespaces {
 		result = append(result, namespace.Name)
@@ -111,7 +115,9 @@ func ListConsulNamespaces(t *testing.T, client *capi.Client) []string {
 // Requires Consul Enterprise.
 func PutConsulKey(t *testing.T, client *capi.Client, namespace, key, value string) {
 	kvClient := client.KV()
-	_, err := kvClient.Put(&capi.KVPair{Key: key, Value: []byte(value)}, &capi.WriteOptions{Namespace: namespace})
+	opts := &capi.WriteOptions{Namespace: namespace}
+
+	_, err := kvClient.Put(&capi.KVPair{Key: key, Value: []byte(value)}, opts)
 	require.NoError(t, err)
 }
 
@@ -120,7 +126,9 @@ func PutConsulKey(t *testing.T, client *capi.Client, namespace, key, value strin
 // Requires Consul Enterprise.
 func DeleteConsulKey(t *testing.T, client *capi.Client, namespace, key string) {
 	kvClient := client.KV()
-	_, err := kvClient.Delete(key, &capi.WriteOptions{Namespace: namespace})
+	opts := &capi.WriteOptions{Namespace: namespace}
+
+	_, err := kvClient.Delete(key, opts)
 	require.NoError(t, err)
 }
 
@@ -130,7 +138,9 @@ func DeleteConsulKey(t *testing.T, client *capi.Client, namespace, key string) {
 // Requires Consul Enterprise.
 func ReadConsulConfigEntry(t *testing.T, client *capi.Client, namespace, kind, name string) capi.ConfigEntry {
 	ceClient := client.ConfigEntries()
-	ce, _, err := ceClient.Get(kind, name, &capi.QueryOptions{Namespace: namespace})
+	opts := &capi.QueryOptions{Namespace: namespace}
+
+	ce, _, err := ceClient.Get(kind, name, opts)
 	require.NoError(t, err)
 	return ce
 }
@@ -141,6 +151,78 @@ func ReadConsulConfigEntry(t *testing.T, client *capi.Client, namespace, kind, n
 // Requires Consul Enterprise.
 func DeleteConsulConfigEntry(t *testing.T, client *capi.Client, namespace, kind, name string) {
 	ceClient := client.ConfigEntries()
-	_, err := ceClient.Delete(kind, name, &capi.WriteOptions{Namespace: namespace})
+	opts := &capi.WriteOptions{Namespace: namespace}
+
+	_, err := ceClient.Delete(kind, name, opts)
 	require.NoError(t, err)
 }
+
+// ConsulPolicy is used for create Consul ACL policies that Consul ACL tokens
+// can make use of.
+type ConsulPolicy struct {
+	Name  string // e.g. nomad-operator
+	Rules string // e.g. service "" { policy="write" }
+}
+
+// CreateConsulPolicy is used to create a Consul ACL policy backed by the given
+// ConsulPolicy in the specified namespace.
+//
+// Requires Consul Enterprise.
+func CreateConsulPolicy(t *testing.T, client *capi.Client, namespace string, policy ConsulPolicy) string {
+	aclClient := client.ACL()
+	opts := &capi.WriteOptions{Namespace: namespace}
+
+	result, _, err := aclClient.PolicyCreate(&capi.ACLPolicy{
+		Name:        policy.Name,
+		Rules:       policy.Rules,
+		Description: fmt.Sprintf("An e2e test policy %q", policy.Name),
+	}, opts)
+	require.NoError(t, err, "failed to create consul acl policy")
+	return result.ID
+}
+
+// DeleteConsulPolicies is used to delete a set Consul ACL policies from Consul.
+//
+// Requires Consul Enterprise.
+func DeleteConsulPolicies(t *testing.T, client *capi.Client, policies map[string][]string) {
+	aclClient := client.ACL()
+
+	for namespace, policyIDs := range policies {
+		opts := &capi.WriteOptions{Namespace: namespace}
+		for _, policyID := range policyIDs {
+			_, err := aclClient.PolicyDelete(policyID, opts)
+			assert.NoError(t, err)
+		}
+	}
+}
+
+// CreateConsulToken is used to create a Consul ACL token backed by the policy of
+// the given policyID in the specified namespace.
+//
+// Requires Consul Enterprise.
+func CreateConsulToken(t *testing.T, client *capi.Client, namespace, policyID string) string {
+	aclClient := client.ACL()
+	opts := &capi.WriteOptions{Namespace: namespace}
+
+	token, _, err := aclClient.TokenCreate(&capi.ACLToken{
+		Policies:    []*capi.ACLTokenPolicyLink{{ID: policyID}},
+		Description: "An e2e test token",
+	}, opts)
+	require.NoError(t, err, "failed to create consul acl token")
+	return token.SecretID
+}
+
+// DeleteConsulTokens is used to delete a set of tokens from Consul.
+//
+// Requires Consul Enterprise.
+func DeleteConsulTokens(t *testing.T, client *capi.Client, tokens map[string][]string) {
+	aclClient := client.ACL()
+
+	for namespace, tokenIDs := range tokens {
+		opts := &capi.WriteOptions{Namespace: namespace}
+		for _, tokenID := range tokenIDs {
+			_, err := aclClient.TokenDelete(tokenID, opts)
+			assert.NoError(t, err)
+		}
+	}
+}
diff --git a/nomad/consul.go b/nomad/consul.go
index 3a8fe56ae50..35a89277e8b 100644
--- a/nomad/consul.go
+++ b/nomad/consul.go
@@ -223,19 +223,25 @@ func (c *consulACLsAPI) CheckPermissions(ctx context.Context, namespace string,
 	}
 
 	// lookup the token from consul
-	token, err := c.readToken(ctx, secretID)
-	if err != nil {
-		return err
+	token, readErr := c.readToken(ctx, secretID)
+	if readErr != nil {
+		return readErr
+	}
+
+	// if the token is a global-management token, it has unrestricted privileges
+	if c.isManagementToken(token) {
+		return nil
 	}
 
-	// verify the token namespace matches namespace in job
-	if token.Namespace != namespace {
-		return errors.Errorf("consul ACL token cannot use namespace %q", namespace)
+	// if the token cannot possibly be used to act on objects in the desired
+	// namespace, reject it immediately
+	if err := namespaceCheck(namespace, token); err != nil {
+		return err
 	}
 
 	// verify token has keystore read permission, if using template
 	if usage.KV {
-		allowable, err := c.canReadKeystore(token)
+		allowable, err := c.canReadKeystore(namespace, token)
 		if err != nil {
 			return err
 		} else if !allowable {
@@ -245,7 +251,7 @@ func (c *consulACLsAPI) CheckPermissions(ctx context.Context, namespace string,
 
 	// verify token has service write permission for group+task services
 	for _, service := range usage.Services {
-		allowable, err := c.canWriteService(service, token)
+		allowable, err := c.canWriteService(namespace, service, token)
 		if err != nil {
 			return err
 		} else if !allowable {
@@ -256,7 +262,7 @@ func (c *consulACLsAPI) CheckPermissions(ctx context.Context, namespace string,
 	// verify token has service identity permission for connect services
 	for _, kind := range usage.Kinds {
 		service := kind.Value()
-		allowable, err := c.canWriteService(service, token)
+		allowable, err := c.canWriteService(namespace, service, token)
 		if err != nil {
 			return err
 		} else if !allowable {
diff --git a/nomad/consul_policy.go b/nomad/consul_policy.go
index b7ba5fecb9a..77bfb47e8dd 100644
--- a/nomad/consul_policy.go
+++ b/nomad/consul_policy.go
@@ -8,6 +8,14 @@ import (
 	"github.com/pkg/errors"
 )
 
+const (
+	// consulGlobalManagementPolicyID is the built-in policy ID used by Consul
+	// to denote global-management tokens.
+	//
+	// https://www.consul.io/docs/security/acl/acl-system#builtin-policies
+	consulGlobalManagementPolicyID = "00000000-0000-0000-0000-000000000001"
+)
+
 // ConsulServiceRule represents a policy for a service.
 type ConsulServiceRule struct {
 	Name   string `hcl:",key"`
@@ -23,41 +31,67 @@ type ConsulKeyRule struct {
 // ConsulPolicy represents the parts of a ConsulServiceRule Policy that are
 // relevant to Service Identity authorizations.
 type ConsulPolicy struct {
-	Services        []*ConsulServiceRule `hcl:"service,expand"`
-	ServicePrefixes []*ConsulServiceRule `hcl:"service_prefix,expand"`
-	KeyPrefixes     []*ConsulKeyRule     `hcl:"key_prefix,expand"`
-}
-
-// IsEmpty returns true if there are no Services, ServicePrefixes, or KeyPrefixes
-// defined for the ConsulPolicy.
-func (cp *ConsulPolicy) IsEmpty() bool {
-	if cp == nil {
-		return true
-	}
-
-	policies := len(cp.Services) + len(cp.ServicePrefixes) + len(cp.KeyPrefixes)
-	return policies == 0
+	Services          []*ConsulServiceRule     `hcl:"service,expand"`
+	ServicePrefixes   []*ConsulServiceRule     `hcl:"service_prefix,expand"`
+	KeyPrefixes       []*ConsulKeyRule         `hcl:"key_prefix,expand"`
+	Namespaces        map[string]*ConsulPolicy `hcl:"namespace,expand"`
+	NamespacePrefixes map[string]*ConsulPolicy `hcl:"namespace_prefix,expand"`
 }
 
-// ParseConsulPolicy parses raw string s into a ConsulPolicy. An error is
+// parseConsulPolicy parses raw string s into a ConsulPolicy. An error is
 // returned if decoding the policy fails, or if the decoded policy has no
 // Services or ServicePrefixes defined.
-func ParseConsulPolicy(s string) (*ConsulPolicy, error) {
+func parseConsulPolicy(s string) (*ConsulPolicy, error) {
 	cp := new(ConsulPolicy)
 	if err := hcl.Decode(cp, s); err != nil {
 		return nil, errors.Wrap(err, "failed to parse ACL policy")
 	}
-	if cp.IsEmpty() {
-		// the only use case for now, may as well validate asap
-		return nil, errors.New("consul policy contains no service rules")
-	}
 	return cp, nil
 }
 
-func (c *consulACLsAPI) canReadKeystore(token *api.ACLToken) (bool, error) {
+// isManagementToken returns true if the Consul token is backed by the
+// built-in global-management policy. Such a token has complete, unrestricted
+// access to all of Consul.
+//
+// https://www.consul.io/docs/security/acl/acl-system#builtin-policies
+func (c *consulACLsAPI) isManagementToken(token *api.ACLToken) bool {
+	if token == nil {
+		return false
+	}
+
+	for _, policy := range token.Policies {
+		if policy.ID == consulGlobalManagementPolicyID {
+			return true
+		}
+	}
+	return false
+}
+
+// namespaceCheck is used to verify the namespace of the object matches the
+// namespace of the ACL token provided.
+//
+// exception: iff token is in the default namespace, it may contain policies
+// that extend into other namespaces using namespace_prefix, which must bypass
+// this early check and validate in the service/keystore helpers
+func namespaceCheck(namespace string, token *api.ACLToken) error {
+	if token.Namespace != "default" && token.Namespace != namespace {
+		return errors.Errorf("consul ACL token cannot use namespace %q", namespace)
+	}
+	return nil
+}
+
+func (c *consulACLsAPI) canReadKeystore(namespace string, token *api.ACLToken) (bool, error) {
+	// early check the token is compatible with desired namespace
+	if err := namespaceCheck(namespace, token); err != nil {
+		return false, nil
+	}
+
+	// determines whether a top-level ACL policy will be applicable
+	matches := namespace == token.Namespace
+
 	// check each policy directly attached to the token
 	for _, policyRef := range token.Policies {
-		if allowable, err := c.policyAllowsKeystoreRead(policyRef.ID); err != nil {
+		if allowable, err := c.policyAllowsKeystoreRead(matches, namespace, policyRef.ID); err != nil {
 			return false, err
 		} else if allowable {
 			return true, nil
@@ -74,7 +108,7 @@ func (c *consulACLsAPI) canReadKeystore(token *api.ACLToken) (bool, error) {
 		}
 
 		for _, policyLink := range role.Policies {
-			allowable, err := c.policyAllowsKeystoreRead(policyLink.ID)
+			allowable, err := c.policyAllowsKeystoreRead(matches, namespace, policyLink.ID)
 			if err != nil {
 				return false, err
 			} else if allowable {
@@ -86,10 +120,18 @@ func (c *consulACLsAPI) canReadKeystore(token *api.ACLToken) (bool, error) {
 	return false, nil
 }
 
-func (c *consulACLsAPI) canWriteService(service string, token *api.ACLToken) (bool, error) {
+func (c *consulACLsAPI) canWriteService(namespace, service string, token *api.ACLToken) (bool, error) {
+	// early check the token is compatible with desired namespace
+	if err := namespaceCheck(namespace, token); err != nil {
+		return false, nil
+	}
+
+	// determines whether a top-level ACL policy will be applicable
+	matches := namespace == token.Namespace
+
 	// check each policy directly attached to the token
 	for _, policyRef := range token.Policies {
-		if allowable, err := c.policyAllowsServiceWrite(service, policyRef.ID); err != nil {
+		if allowable, err := c.policyAllowsServiceWrite(matches, namespace, service, policyRef.ID); err != nil {
 			return false, err
 		} else if allowable {
 			return true, nil
@@ -106,9 +148,9 @@ func (c *consulACLsAPI) canWriteService(service string, token *api.ACLToken) (bo
 		}
 
 		for _, policyLink := range role.Policies {
-			allowable, err := c.policyAllowsServiceWrite(service, policyLink.ID)
-			if err != nil {
-				return false, err
+			allowable, wErr := c.policyAllowsServiceWrite(matches, namespace, service, policyLink.ID)
+			if wErr != nil {
+				return false, wErr
 			} else if allowable {
 				return true, nil
 			}
@@ -118,7 +160,7 @@ func (c *consulACLsAPI) canWriteService(service string, token *api.ACLToken) (bo
 	return false, nil
 }
 
-func (c *consulACLsAPI) policyAllowsServiceWrite(service string, policyID string) (bool, error) {
+func (c *consulACLsAPI) policyAllowsServiceWrite(matches bool, namespace, service string, policyID string) (bool, error) {
 	policy, _, err := c.aclClient.PolicyRead(policyID, &api.QueryOptions{
 		AllowStale: false,
 	})
@@ -129,15 +171,14 @@ func (c *consulACLsAPI) policyAllowsServiceWrite(service string, policyID string
 	// compare policy to the necessary permission for service write
 	// e.g. service "db" { policy = "write" }
 	// e.g. service_prefix "" { policy == "write" }
-	cp, err := ParseConsulPolicy(policy.Rules)
+	cp, err := parseConsulPolicy(policy.Rules)
 	if err != nil {
 		return false, err
 	}
 
-	if cp.allowsServiceWrite(service) {
+	if cp.allowsServiceWrite(matches, namespace, service) {
 		return true, nil
 	}
-
 	return false, nil
 }
 
@@ -145,30 +186,65 @@ const (
 	serviceNameWildcard = "*"
 )
 
-func (cp *ConsulPolicy) allowsServiceWrite(task string) bool {
-	for _, service := range cp.Services {
-		name := strings.ToLower(service.Name)
-		policy := strings.ToLower(service.Policy)
-		if policy == ConsulPolicyWrite {
-			if name == task || name == serviceNameWildcard {
+func (cp *ConsulPolicy) allowsServiceWrite(matches bool, namespace, task string) bool {
+	canWriteService := func(services []*ConsulServiceRule) bool {
+		for _, service := range services {
+			name := strings.ToLower(service.Name)
+			policy := strings.ToLower(service.Policy)
+			if policy == ConsulPolicyWrite {
+				if name == task || name == serviceNameWildcard {
+					return true
+				}
+			}
+		}
+		return false
+	}
+
+	canWriteServicePrefix := func(services []*ConsulServiceRule) bool {
+		for _, servicePrefix := range services {
+			prefix := strings.ToLower(servicePrefix.Name)
+			policy := strings.ToLower(servicePrefix.Policy)
+			if policy == ConsulPolicyWrite {
+				if strings.HasPrefix(task, prefix) {
+					return true
+				}
+			}
+		}
+		return false
+	}
+
+	if matches {
+		// check the top-level service/service_prefix rules
+		if canWriteService(cp.Services) || canWriteServicePrefix(cp.ServicePrefixes) {
+			return true
+		}
+	}
+
+	// for each namespace rule, if that namespace and the desired namespace
+	// are a match, we can then check the service/service_prefix policy rules
+	for ns, policy := range cp.Namespaces {
+		if ns == namespace {
+			if canWriteService(policy.Services) || canWriteServicePrefix(policy.ServicePrefixes) {
 				return true
 			}
 		}
 	}
 
-	for _, servicePrefix := range cp.ServicePrefixes {
-		prefix := strings.ToLower(servicePrefix.Name)
-		policy := strings.ToLower(servicePrefix.Policy)
-		if policy == ConsulPolicyWrite {
-			if strings.HasPrefix(task, prefix) {
+	// for each namespace_prefix rule, see if that namespace_prefix applies
+	// to this namespace, and if yes, also check those service/service_prefix
+	// policy rules
+	for prefix, policy := range cp.NamespacePrefixes {
+		if strings.HasPrefix(namespace, prefix) {
+			if canWriteService(policy.Services) || canWriteServicePrefix(policy.ServicePrefixes) {
 				return true
 			}
 		}
 	}
+
 	return false
 }
 
-func (c *consulACLsAPI) policyAllowsKeystoreRead(policyID string) (bool, error) {
+func (c *consulACLsAPI) policyAllowsKeystoreRead(matches bool, namespace, policyID string) (bool, error) {
 	policy, _, err := c.aclClient.PolicyRead(policyID, &api.QueryOptions{
 		AllowStale: false,
 	})
@@ -176,27 +252,57 @@ func (c *consulACLsAPI) policyAllowsKeystoreRead(policyID string) (bool, error)
 		return false, err
 	}
 
-	cp, err := ParseConsulPolicy(policy.Rules)
+	cp, err := parseConsulPolicy(policy.Rules)
 	if err != nil {
 		return false, err
 	}
 
-	if cp.allowsKeystoreRead() {
+	if cp.allowsKeystoreRead(matches, namespace) {
 		return true, nil
 	}
 
 	return false, nil
 }
 
-func (cp *ConsulPolicy) allowsKeystoreRead() bool {
-	for _, keyPrefix := range cp.KeyPrefixes {
-		name := strings.ToLower(keyPrefix.Name)
-		policy := strings.ToLower(keyPrefix.Policy)
-		if name == "" {
-			if policy == ConsulPolicyWrite || policy == ConsulPolicyRead {
+func (cp *ConsulPolicy) allowsKeystoreRead(matches bool, namespace string) bool {
+	canReadKeystore := func(prefixes []*ConsulKeyRule) bool {
+		for _, keyPrefix := range prefixes {
+			name := strings.ToLower(keyPrefix.Name)
+			policy := strings.ToLower(keyPrefix.Policy)
+			if name == "" {
+				if policy == ConsulPolicyWrite || policy == ConsulPolicyRead {
+					return true
+				}
+			}
+		}
+		return false
+	}
+
+	// check the top-level key_prefix rules, but only if the desired namespace
+	// matches the namespace of the consul acl token
+	if matches && canReadKeystore(cp.KeyPrefixes) {
+		return true
+	}
+
+	// for each namespace rule, if that namespace matches the desired namespace
+	// we chan then check the keystore policy
+	for ns, policy := range cp.Namespaces {
+		if ns == namespace {
+			if canReadKeystore(policy.KeyPrefixes) {
 				return true
 			}
 		}
 	}
+
+	// for each namespace_prefix rule, see if that namespace_prefix applies to
+	// this namespace, and if yes, also check those key_prefix policy rules
+	for prefix, policy := range cp.NamespacePrefixes {
+		if strings.HasPrefix(namespace, prefix) {
+			if canReadKeystore(policy.KeyPrefixes) {
+				return true
+			}
+		}
+	}
+
 	return false
 }
diff --git a/nomad/consul_policy_test.go b/nomad/consul_policy_test.go
index a38d57e60f8..2764b49679a 100644
--- a/nomad/consul_policy_test.go
+++ b/nomad/consul_policy_test.go
@@ -6,6 +6,7 @@ import (
 	"github.com/hashicorp/consul/api"
 	"github.com/hashicorp/nomad/command/agent/consul"
 	"github.com/hashicorp/nomad/helper/testlog"
+	"github.com/hashicorp/nomad/helper/uuid"
 	"github.com/stretchr/testify/require"
 )
 
@@ -13,10 +14,9 @@ func TestConsulPolicy_ParseConsulPolicy(t *testing.T) {
 	t.Parallel()
 
 	try := func(t *testing.T, text string, expPolicy *ConsulPolicy, expErr string) {
-		policy, err := ParseConsulPolicy(text)
+		policy, err := parseConsulPolicy(text)
 		if expErr != "" {
 			require.EqualError(t, err, expErr)
-			require.True(t, policy.IsEmpty())
 		} else {
 			require.NoError(t, err)
 			require.Equal(t, expPolicy, policy)
@@ -26,8 +26,7 @@ func TestConsulPolicy_ParseConsulPolicy(t *testing.T) {
 	t.Run("service", func(t *testing.T) {
 		text := `service "web" { policy = "read" }`
 		exp := &ConsulPolicy{
-			Services:        []*ConsulServiceRule{{Name: "web", Policy: "read"}},
-			ServicePrefixes: []*ConsulServiceRule(nil),
+			Services: []*ConsulServiceRule{{Name: "web", Policy: "read"}},
 		}
 		try(t, text, exp, "")
 	})
@@ -35,16 +34,17 @@ func TestConsulPolicy_ParseConsulPolicy(t *testing.T) {
 	t.Run("service_prefix", func(t *testing.T) {
 		text := `service_prefix "data" { policy = "write" }`
 		exp := &ConsulPolicy{
-			Services:        []*ConsulServiceRule(nil),
 			ServicePrefixes: []*ConsulServiceRule{{Name: "data", Policy: "write"}},
 		}
 		try(t, text, exp, "")
 	})
 
-	t.Run("empty", func(t *testing.T) {
-		text := ``
-		expErr := "consul policy contains no service rules"
-		try(t, text, nil, expErr)
+	t.Run("key_prefix", func(t *testing.T) {
+		text := `key_prefix "keys" { policy = "read" }`
+		exp := &ConsulPolicy{
+			KeyPrefixes: []*ConsulKeyRule{{Name: "keys", Policy: "read"}},
+		}
+		try(t, text, exp, "")
 	})
 
 	t.Run("malformed", func(t *testing.T) {
@@ -52,53 +52,75 @@ func TestConsulPolicy_ParseConsulPolicy(t *testing.T) {
 		expErr := "failed to parse ACL policy: At 1:22: illegal char"
 		try(t, text, nil, expErr)
 	})
-}
 
-func TestConsulPolicy_IsEmpty(t *testing.T) {
-	t.Parallel()
+	t.Run("multi-namespace", func(t *testing.T) {
+		text := `
+service_prefix "z" { policy = "write" }
 
-	try := func(t *testing.T, cp *ConsulPolicy, exp bool) {
-		result := cp.IsEmpty()
-		require.Equal(t, exp, result)
-	}
-
-	t.Run("nil", func(t *testing.T) {
-		cp := (*ConsulPolicy)(nil)
-		try(t, cp, true)
-	})
+namespace_prefix "b" {
+  service_prefix "b" { policy = "write" }
+  key_prefix "" { policy = "read" }
+}
 
-	t.Run("empty slices", func(t *testing.T) {
-		cp := &ConsulPolicy{
-			Services:        []*ConsulServiceRule(nil),
-			ServicePrefixes: []*ConsulServiceRule(nil),
-		}
-		try(t, cp, true)
-	})
+namespace_prefix "c" {
+  service_prefix "c" { policy = "read" }
+  key_prefix "" { policy = "read" }
+}
 
-	t.Run("services nonempty", func(t *testing.T) {
-		cp := &ConsulPolicy{
-			Services: []*ConsulServiceRule{{Name: "example", Policy: "write"}},
-		}
-		try(t, cp, false)
-	})
+namespace_prefix "" {
+  key_prefix "shared/" { policy = "read" }
+}
 
-	t.Run("service_prefixes nonempty", func(t *testing.T) {
-		cp := &ConsulPolicy{
-			ServicePrefixes: []*ConsulServiceRule{{Name: "pre", Policy: "read"}},
+namespace "foo" {
+  service "bar" { policy = "read" }
+  service_prefix "foo-" { policy = "write" }
+  key_prefix "" { policy = "read" }
+}
+`
+		exp := &ConsulPolicy{
+			ServicePrefixes: []*ConsulServiceRule{{Name: "z", Policy: "write"}},
+			NamespacePrefixes: map[string]*ConsulPolicy{
+				"b": {
+					ServicePrefixes: []*ConsulServiceRule{{Name: "b", Policy: "write"}},
+					KeyPrefixes:     []*ConsulKeyRule{{Name: "", Policy: "read"}},
+				},
+				"c": {
+					ServicePrefixes: []*ConsulServiceRule{{Name: "c", Policy: "read"}},
+					KeyPrefixes:     []*ConsulKeyRule{{Name: "", Policy: "read"}},
+				},
+				"": {
+					KeyPrefixes: []*ConsulKeyRule{{Name: "shared/", Policy: "read"}},
+				},
+			},
+			Namespaces: map[string]*ConsulPolicy{
+				"foo": {
+					Services:        []*ConsulServiceRule{{Name: "bar", Policy: "read"}},
+					ServicePrefixes: []*ConsulServiceRule{{Name: "foo-", Policy: "write"}},
+					KeyPrefixes:     []*ConsulKeyRule{{Name: "", Policy: "read"}},
+				},
+			},
 		}
-		try(t, cp, false)
+		try(t, text, exp, "")
 	})
 }
 
 func TestConsulACLsAPI_allowsServiceWrite(t *testing.T) {
 	t.Parallel()
 
-	try := func(t *testing.T, task string, cp *ConsulPolicy, exp bool) {
-		result := cp.allowsServiceWrite(task)
+	try := func(t *testing.T, matches bool, namespace, task string, cp *ConsulPolicy, exp bool) {
+		// If matches is false, the implication is that the consul acl token is in
+		// the default namespace, otherwise prior validation would stop the request
+		// before getting to policy checks. Only consul acl tokens in the default
+		// namespace are allowed to have namespace_prefix blocks.
+		result := cp.allowsServiceWrite(matches, namespace, task)
 		require.Equal(t, exp, result)
 	}
 
-	makeCP := func(services [][2]string, prefixes [][2]string) *ConsulPolicy {
+	// create a consul policy backed by service and/or service_prefix rules
+	//
+	// if namespace == "_", use the top level service/service_prefix rules, otherwise
+	// set the rules as a namespace_prefix ruleset
+	makeCP := func(namespace string, services [][2]string, prefixes [][2]string) *ConsulPolicy {
 		serviceRules := make([]*ConsulServiceRule, 0, len(services))
 		for _, service := range services {
 			serviceRules = append(serviceRules, &ConsulServiceRule{Name: service[0], Policy: service[1]})
@@ -107,147 +129,526 @@ func TestConsulACLsAPI_allowsServiceWrite(t *testing.T) {
 		for _, prefix := range prefixes {
 			prefixRules = append(prefixRules, &ConsulServiceRule{Name: prefix[0], Policy: prefix[1]})
 		}
-		return &ConsulPolicy{Services: serviceRules, ServicePrefixes: prefixRules}
+
+		if namespace == "_" {
+			return &ConsulPolicy{Services: serviceRules, ServicePrefixes: prefixRules}
+		}
+
+		return &ConsulPolicy{
+			Namespaces: map[string]*ConsulPolicy{
+				namespace: {
+					Services:        serviceRules,
+					ServicePrefixes: prefixRules,
+				},
+			},
+			NamespacePrefixes: map[string]*ConsulPolicy{
+				namespace: {
+					Services:        serviceRules,
+					ServicePrefixes: prefixRules,
+				},
+			}}
 	}
 
 	t.Run("matching service policy write", func(t *testing.T) {
-		try(t, "task1", makeCP(
-			[][2]string{{"task1", "write"}},
-			nil,
-		), true)
+		rule := [][2]string{{"task1", "write"}}
+		const task = "task1"
+		t.Run("namespaces match", func(t *testing.T) {
+			const matches = true
+			try(t, matches, "default", task, makeCP("_", rule, nil), true)
+			try(t, matches, "default", task, makeCP("default", rule, nil), true)
+			try(t, matches, "apple", task, makeCP("_", rule, nil), true)
+			try(t, matches, "apple", task, makeCP("apple", rule, nil), true)
+			try(t, matches, "apple", task, makeCP("app", rule, nil), true)
+			try(t, matches, "other", task, makeCP("", rule, nil), true)
+			try(t, matches, "other", task, makeCP("apple", rule, nil), false)
+		})
+		t.Run("namespaces do not match", func(t *testing.T) {
+			const matches = false
+			try(t, matches, "apple", task, makeCP("_", rule, nil), false)
+			try(t, matches, "apple", task, makeCP("apple", rule, nil), true)
+			try(t, matches, "apple", task, makeCP("app", rule, nil), true)
+			try(t, matches, "other", task, makeCP("", rule, nil), true)
+			try(t, matches, "other", task, makeCP("apple", rule, nil), false)
+		})
 	})
 
 	t.Run("matching service policy read", func(t *testing.T) {
-		try(t, "task1", makeCP(
-			[][2]string{{"task1", "read"}},
-			nil,
-		), false)
+		rule := [][2]string{{"task1", "read"}}
+		const task = "task1"
+		t.Run("namespaces match", func(t *testing.T) {
+			const matches = true
+			try(t, matches, "default", task, makeCP("_", rule, nil), false)
+			try(t, matches, "default", task, makeCP("default", rule, nil), false)
+			try(t, matches, "apple", task, makeCP("_", rule, nil), false)
+			try(t, matches, "apple", task, makeCP("apple", rule, nil), false)
+			try(t, matches, "apple", task, makeCP("app", rule, nil), false)
+			try(t, matches, "other", task, makeCP("", rule, nil), false)
+			try(t, matches, "other", task, makeCP("apple", rule, nil), false)
+		})
+		t.Run("namespaces do not match", func(t *testing.T) {
+			const matches = false
+			try(t, matches, "apple", task, makeCP("_", rule, nil), false)
+			try(t, matches, "apple", task, makeCP("apple", rule, nil), false)
+			try(t, matches, "apple", task, makeCP("app", rule, nil), false)
+			try(t, matches, "other", task, makeCP("", rule, nil), false)
+			try(t, matches, "other", task, makeCP("apple", rule, nil), false)
+		})
 	})
 
 	t.Run("wildcard service policy write", func(t *testing.T) {
-		try(t, "task1", makeCP(
-			[][2]string{{"*", "write"}},
-			nil,
-		), true)
+		rule := [][2]string{{"*", "write"}}
+		const task = "task1"
+		t.Run("namespaces match", func(t *testing.T) {
+			const matches = true
+			try(t, matches, "default", task, makeCP("_", rule, nil), true)
+			try(t, matches, "default", task, makeCP("default", rule, nil), true)
+			try(t, matches, "apple", task, makeCP("_", rule, nil), true)
+			try(t, matches, "apple", task, makeCP("app", rule, nil), true)
+			try(t, matches, "other", task, makeCP("", rule, nil), true)
+			try(t, matches, "other", task, makeCP("apple", rule, nil), false)
+		})
+		t.Run("namespaces do not match", func(t *testing.T) {
+			const matches = false
+			try(t, matches, "apple", task, makeCP("_", rule, nil), false)
+			try(t, matches, "apple", task, makeCP("app", rule, nil), true)
+			try(t, matches, "other", task, makeCP("", rule, nil), true)
+			try(t, matches, "other", task, makeCP("apple", rule, nil), false)
+		})
 	})
 
 	t.Run("wrong service policy write", func(t *testing.T) {
-		try(t, "other1", makeCP(
-			[][2]string{{"task1", "write"}},
-			nil,
-		), false)
+		rule := [][2]string{{"task1", "write"}}
+		const task = "other1"
+		t.Run("namespaces match", func(t *testing.T) {
+			const matches = true
+			try(t, matches, "default", task, makeCP("_", rule, nil), false)
+			try(t, matches, "default", task, makeCP("default", rule, nil), false)
+			try(t, matches, "apple", task, makeCP("_", rule, nil), false)
+			try(t, matches, "apple", task, makeCP("app", rule, nil), false)
+			try(t, matches, "other", task, makeCP("", rule, nil), false)
+			try(t, matches, "other", task, makeCP("apple", rule, nil), false)
+		})
+		t.Run("namespaces do not match", func(t *testing.T) {
+			const matches = true
+			try(t, matches, "apple", task, makeCP("_", rule, nil), false)
+			try(t, matches, "apple", task, makeCP("app", rule, nil), false)
+			try(t, matches, "other", task, makeCP("", rule, nil), false)
+			try(t, matches, "other", task, makeCP("apple", rule, nil), false)
+		})
 	})
 
 	t.Run("matching prefix policy write", func(t *testing.T) {
-		try(t, "task-one", makeCP(
-			nil,
-			[][2]string{{"task-", "write"}},
-		), true)
+		rule := [][2]string{{"task-", "write"}}
+		const task = "task-one"
+		t.Run("namespaces match", func(t *testing.T) {
+			const matches = true
+			try(t, matches, "default", task, makeCP("_", nil, rule), true)
+			try(t, matches, "default", task, makeCP("default", nil, rule), true)
+			try(t, matches, "apple", task, makeCP("_", nil, rule), true)
+			try(t, matches, "apple", task, makeCP("app", nil, rule), true)
+			try(t, matches, "other", task, makeCP("", nil, rule), true)
+			try(t, matches, "other", task, makeCP("apple", nil, rule), false)
+		})
+		t.Run("namespaces do not match", func(t *testing.T) {
+			const matches = false
+			try(t, matches, "apple", task, makeCP("_", nil, rule), false)
+			try(t, matches, "apple", task, makeCP("app", nil, rule), true)
+			try(t, matches, "other", task, makeCP("", nil, rule), true)
+			try(t, matches, "other", task, makeCP("apple", nil, rule), false)
+		})
 	})
 
 	t.Run("matching prefix policy read", func(t *testing.T) {
-		try(t, "task-one", makeCP(
-			nil,
-			[][2]string{{"task-", "read"}},
-		), false)
+		rule := [][2]string{{"task-", "read"}}
+		const task = "task-one"
+		t.Run("namespaces match", func(t *testing.T) {
+			const matches = true
+			try(t, matches, "default", task, makeCP("_", nil, rule), false)
+			try(t, matches, "default", task, makeCP("default", nil, rule), false)
+			try(t, matches, "apple", task, makeCP("_", nil, rule), false)
+			try(t, matches, "apple", task, makeCP("app", nil, rule), false)
+			try(t, matches, "other", task, makeCP("", nil, rule), false)
+			try(t, matches, "other", task, makeCP("apple", nil, rule), false)
+		})
+		t.Run("namespaces do not match", func(t *testing.T) {
+			const matches = false
+			try(t, matches, "apple", task, makeCP("_", nil, rule), false)
+			try(t, matches, "apple", task, makeCP("app", nil, rule), false)
+			try(t, matches, "other", task, makeCP("", nil, rule), false)
+			try(t, matches, "other", task, makeCP("apple", nil, rule), false)
+		})
 	})
 
 	t.Run("empty prefix policy write", func(t *testing.T) {
-		try(t, "task-one", makeCP(
-			nil,
-			[][2]string{{"", "write"}},
-		), true)
+		rule := [][2]string{{"", "write"}}
+		const task = "task-one"
+		t.Run("namespaces match", func(t *testing.T) {
+			const matches = true
+			try(t, matches, "default", task, makeCP("_", nil, rule), true)
+			try(t, matches, "default", task, makeCP("default", nil, rule), true)
+			try(t, matches, "apple", task, makeCP("_", nil, rule), true)
+			try(t, matches, "apple", task, makeCP("app", nil, rule), true)
+			try(t, matches, "other", task, makeCP("", nil, rule), true)
+			try(t, matches, "other", task, makeCP("apple", nil, rule), false)
+		})
+		t.Run("namespaces do not match", func(t *testing.T) {
+			const matches = false
+			try(t, matches, "apple", task, makeCP("_", nil, rule), false)
+			try(t, matches, "apple", task, makeCP("app", nil, rule), true)
+			try(t, matches, "other", task, makeCP("", nil, rule), true)
+			try(t, matches, "other", task, makeCP("apple", nil, rule), false)
+		})
 	})
 
 	t.Run("late matching service", func(t *testing.T) {
-		try(t, "task1", makeCP(
-			[][2]string{{"task0", "write"}, {"task1", "write"}},
-			nil,
-		), true)
+		rule := [][2]string{{"task0", "write"}, {"task1", "write"}}
+		const task = "task1"
+		t.Run("namespaces match", func(t *testing.T) {
+			const matches = true
+			try(t, matches, "default", task, makeCP("_", rule, nil), true)
+			try(t, matches, "default", task, makeCP("default", rule, nil), true)
+			try(t, matches, "apple", task, makeCP("_", rule, nil), true)
+			try(t, matches, "apple", task, makeCP("app", rule, nil), true)
+			try(t, matches, "other", task, makeCP("", rule, nil), true)
+			try(t, matches, "other", task, makeCP("apple", rule, nil), false)
+		})
+		t.Run("namespaces do not match", func(t *testing.T) {
+			const matches = false
+			try(t, matches, "apple", task, makeCP("_", rule, nil), false)
+			try(t, matches, "apple", task, makeCP("app", rule, nil), true)
+			try(t, matches, "other", task, makeCP("", rule, nil), true)
+			try(t, matches, "other", task, makeCP("apple", rule, nil), false)
+		})
 	})
 
 	t.Run("late matching prefix", func(t *testing.T) {
-		try(t, "task-one", makeCP(
-			nil,
-			[][2]string{{"foo-", "write"}, {"task-", "write"}},
-		), true)
+		rule := [][2]string{{"foo-", "write"}, {"task-", "write"}}
+		const task = "task-one"
+		t.Run("namespaces match", func(t *testing.T) {
+			const matches = true
+			try(t, matches, "default", task, makeCP("_", nil, rule), true)
+			try(t, matches, "default", task, makeCP("default", nil, rule), true)
+			try(t, matches, "apple", task, makeCP("_", nil, rule), true)
+			try(t, matches, "apple", task, makeCP("app", nil, rule), true)
+			try(t, matches, "other", task, makeCP("", nil, rule), true)
+			try(t, matches, "other", task, makeCP("apple", nil, rule), false)
+		})
+		t.Run("namespaces do not match", func(t *testing.T) {
+			const matches = false
+			try(t, matches, "apple", task, makeCP("_", nil, rule), false)
+			try(t, matches, "apple", task, makeCP("app", nil, rule), true)
+			try(t, matches, "other", task, makeCP("", nil, rule), true)
+			try(t, matches, "other", task, makeCP("apple", nil, rule), false)
+		})
 	})
 }
 
 func TestConsulACLsAPI_hasSufficientPolicy(t *testing.T) {
 	t.Parallel()
 
-	try := func(t *testing.T, task string, token *api.ACLToken, exp bool) {
+	try := func(t *testing.T, namespace, task string, token *api.ACLToken, exp bool) {
 		logger := testlog.HCLogger(t)
 		cAPI := &consulACLsAPI{
 			aclClient: consul.NewMockACLsAPI(logger),
 			logger:    logger,
 		}
-		result, err := cAPI.canWriteService(task, token)
+		result, err := cAPI.canWriteService(namespace, task, token)
 		require.NoError(t, err)
 		require.Equal(t, exp, result)
 	}
 
-	t.Run("no useful policy or role", func(t *testing.T) {
-		try(t, "service1", consul.ExampleOperatorToken0, false)
+	t.Run("default namespace with default token", func(t *testing.T) {
+		t.Run("no useful policy or role", func(t *testing.T) {
+			try(t, "default", "service1", consul.ExampleOperatorToken0, false)
+		})
+
+		t.Run("working policy only", func(t *testing.T) {
+			try(t, "default", "service1", consul.ExampleOperatorToken1, true)
+		})
+
+		t.Run("working role only", func(t *testing.T) {
+			try(t, "default", "service1", consul.ExampleOperatorToken4, true)
+		})
 	})
 
-	t.Run("working policy only", func(t *testing.T) {
-		try(t, "service1", consul.ExampleOperatorToken1, true)
+	t.Run("other namespace with default token", func(t *testing.T) {
+		t.Run("no useful policy or role", func(t *testing.T) {
+			try(t, "other", "service1", consul.ExampleOperatorToken0, false)
+		})
+
+		t.Run("working policy only", func(t *testing.T) {
+			try(t, "other", "service1", consul.ExampleOperatorToken1, false)
+		})
+
+		t.Run("working role only", func(t *testing.T) {
+			try(t, "other", "service1", consul.ExampleOperatorToken4, false)
+		})
 	})
 
-	t.Run("working role only", func(t *testing.T) {
-		try(t, "service1", consul.ExampleOperatorToken4, true)
+	t.Run("default namespace with banana token", func(t *testing.T) {
+		t.Run("no useful policy or role", func(t *testing.T) {
+			try(t, "default", "service1", consul.ExampleOperatorToken10, false)
+		})
+
+		t.Run("working policy only", func(t *testing.T) {
+			try(t, "default", "service1", consul.ExampleOperatorToken11, false)
+		})
+
+		t.Run("working role only", func(t *testing.T) {
+			try(t, "default", "service1", consul.ExampleOperatorToken14, false)
+		})
+	})
+
+	t.Run("banana namespace with banana token", func(t *testing.T) {
+		t.Run("no useful policy or role", func(t *testing.T) {
+			try(t, "banana", "service1", consul.ExampleOperatorToken10, false)
+		})
+
+		t.Run("working policy only", func(t *testing.T) {
+			try(t, "banana", "service1", consul.ExampleOperatorToken11, true)
+		})
+
+		t.Run("working role only", func(t *testing.T) {
+			try(t, "banana", "service1", consul.ExampleOperatorToken14, true)
+		})
 	})
 }
 
 func TestConsulPolicy_allowKeystoreRead(t *testing.T) {
 	t.Run("empty", func(t *testing.T) {
-		require.False(t, new(ConsulPolicy).allowsKeystoreRead())
+		require.False(t, new(ConsulPolicy).allowsKeystoreRead(true, "default"))
 	})
 
 	t.Run("services only", func(t *testing.T) {
-		require.False(t, (&ConsulPolicy{
+		policy := &ConsulPolicy{
 			Services: []*ConsulServiceRule{{
 				Name:   "service1",
 				Policy: "write",
 			}},
-		}).allowsKeystoreRead())
+		}
+		require.False(t, policy.allowsKeystoreRead(true, "default"))
+		require.False(t, policy.allowsKeystoreRead(false, "apple"))
 	})
 
+	// using top-level key_prefix block
+
 	t.Run("kv any read", func(t *testing.T) {
-		require.True(t, (&ConsulPolicy{
+		policy := &ConsulPolicy{
 			KeyPrefixes: []*ConsulKeyRule{{
 				Name:   "",
 				Policy: "read",
 			}},
-		}).allowsKeystoreRead())
+		}
+		require.True(t, policy.allowsKeystoreRead(true, "default"))
+		require.False(t, policy.allowsKeystoreRead(false, "apple"))
 	})
 
 	t.Run("kv any write", func(t *testing.T) {
-		require.True(t, (&ConsulPolicy{
+		policy := &ConsulPolicy{
 			KeyPrefixes: []*ConsulKeyRule{{
 				Name:   "",
 				Policy: "write",
 			}},
-		}).allowsKeystoreRead())
+		}
+		require.True(t, policy.allowsKeystoreRead(true, "default"))
+		require.False(t, policy.allowsKeystoreRead(false, "apple"))
 	})
 
 	t.Run("kv limited read", func(t *testing.T) {
-		require.False(t, (&ConsulPolicy{
+		policy := &ConsulPolicy{
 			KeyPrefixes: []*ConsulKeyRule{{
 				Name:   "foo/bar",
 				Policy: "read",
 			}},
-		}).allowsKeystoreRead())
+		}
+		require.False(t, policy.allowsKeystoreRead(true, "default"))
+		require.False(t, policy.allowsKeystoreRead(false, "apple"))
 	})
 
 	t.Run("kv limited write", func(t *testing.T) {
-		require.False(t, (&ConsulPolicy{
+		policy := &ConsulPolicy{
 			KeyPrefixes: []*ConsulKeyRule{{
 				Name:   "foo/bar",
 				Policy: "write",
 			}},
-		}).allowsKeystoreRead())
+		}
+		require.False(t, policy.allowsKeystoreRead(true, "default"))
+		require.False(t, policy.allowsKeystoreRead(false, "apple"))
+	})
+
+	// using namespace_prefix block
+
+	t.Run("kv wild namespace prefix any read", func(t *testing.T) {
+		policy := &ConsulPolicy{
+			NamespacePrefixes: map[string]*ConsulPolicy{
+				"": &ConsulPolicy{
+					KeyPrefixes: []*ConsulKeyRule{{
+						Name:   "",
+						Policy: "read",
+					}},
+				},
+			},
+		}
+		require.True(t, policy.allowsKeystoreRead(true, "default"))
+		require.True(t, policy.allowsKeystoreRead(false, "apple"))
+	})
+
+	t.Run("kv apple namespace prefix any read", func(t *testing.T) {
+		policy := &ConsulPolicy{
+			NamespacePrefixes: map[string]*ConsulPolicy{
+				"apple": &ConsulPolicy{
+					KeyPrefixes: []*ConsulKeyRule{{
+						Name:   "",
+						Policy: "read",
+					}},
+				},
+			},
+		}
+		require.False(t, policy.allowsKeystoreRead(true, "default"))
+		require.True(t, policy.allowsKeystoreRead(false, "apple"))
+	})
+
+	t.Run("kv matching namespace prefix any read", func(t *testing.T) {
+		policy := &ConsulPolicy{
+			NamespacePrefixes: map[string]*ConsulPolicy{
+				"app": &ConsulPolicy{
+					KeyPrefixes: []*ConsulKeyRule{{
+						Name:   "",
+						Policy: "read",
+					}},
+				},
+			},
+		}
+		require.False(t, policy.allowsKeystoreRead(true, "default"))
+		require.True(t, policy.allowsKeystoreRead(false, "apple"))
+	})
+
+	t.Run("kv other namespace prefix any read", func(t *testing.T) {
+		policy := &ConsulPolicy{
+			NamespacePrefixes: map[string]*ConsulPolicy{
+				"other": &ConsulPolicy{
+					KeyPrefixes: []*ConsulKeyRule{{
+						Name:   "",
+						Policy: "read",
+					}},
+				},
+			},
+		}
+		require.False(t, policy.allowsKeystoreRead(true, "default"))
+		require.False(t, policy.allowsKeystoreRead(false, "apple"))
+	})
+
+	// using namespace block
+
+	t.Run("kv match namespace any read", func(t *testing.T) {
+		policy := &ConsulPolicy{
+			Namespaces: map[string]*ConsulPolicy{
+				"apple": &ConsulPolicy{
+					KeyPrefixes: []*ConsulKeyRule{{
+						Name:   "",
+						Policy: "read",
+					}},
+				},
+			},
+		}
+		require.False(t, policy.allowsKeystoreRead(true, "default"))
+		require.True(t, policy.allowsKeystoreRead(true, "apple"))
+	})
+
+	t.Run("kv mismatch namespace any read", func(t *testing.T) {
+		policy := &ConsulPolicy{
+			Namespaces: map[string]*ConsulPolicy{
+				"other": &ConsulPolicy{
+					KeyPrefixes: []*ConsulKeyRule{{
+						Name:   "",
+						Policy: "read",
+					}},
+				},
+			},
+		}
+		require.False(t, policy.allowsKeystoreRead(true, "default"))
+		require.False(t, policy.allowsKeystoreRead(true, "apple"))
+	})
+
+	t.Run("kv matching namespace prefix any read", func(t *testing.T) {
+		policy := &ConsulPolicy{
+			Namespaces: map[string]*ConsulPolicy{
+				"apple": &ConsulPolicy{
+					KeyPrefixes: []*ConsulKeyRule{{
+						Name:   "",
+						Policy: "read",
+					}},
+				},
+			},
+		}
+		require.False(t, policy.allowsKeystoreRead(false, "default"))
+		require.True(t, policy.allowsKeystoreRead(false, "apple"))
+	})
+
+	t.Run("kv mismatch namespace prefix any read", func(t *testing.T) {
+		policy := &ConsulPolicy{
+			Namespaces: map[string]*ConsulPolicy{
+				"other": &ConsulPolicy{
+					KeyPrefixes: []*ConsulKeyRule{{
+						Name:   "",
+						Policy: "read",
+					}},
+				},
+			},
+		}
+		require.False(t, policy.allowsKeystoreRead(true, "default"))
+		require.False(t, policy.allowsKeystoreRead(true, "apple"))
+	})
+}
+
+func TestConsulPolicy_isManagementToken(t *testing.T) {
+	aclsAPI := new(consulACLsAPI)
+
+	t.Run("nil", func(t *testing.T) {
+		token := (*api.ACLToken)(nil)
+		result := aclsAPI.isManagementToken(token)
+		require.False(t, result)
+	})
+
+	t.Run("no policies", func(t *testing.T) {
+		token := &api.ACLToken{
+			Policies: []*api.ACLTokenPolicyLink{},
+		}
+		result := aclsAPI.isManagementToken(token)
+		require.False(t, result)
+	})
+
+	t.Run("management policy", func(t *testing.T) {
+		token := &api.ACLToken{
+			Policies: []*api.ACLTokenPolicyLink{{
+				ID: consulGlobalManagementPolicyID,
+			}},
+		}
+		result := aclsAPI.isManagementToken(token)
+		require.True(t, result)
+	})
+
+	t.Run("other policy", func(t *testing.T) {
+		token := &api.ACLToken{
+			Policies: []*api.ACLTokenPolicyLink{{
+				ID: uuid.Generate(),
+			}},
+		}
+		result := aclsAPI.isManagementToken(token)
+		require.False(t, result)
+	})
+
+	t.Run("mixed policies", func(t *testing.T) {
+		token := &api.ACLToken{
+			Policies: []*api.ACLTokenPolicyLink{{
+				ID: uuid.Generate(),
+			}, {
+				ID: consulGlobalManagementPolicyID,
+			}, {
+				ID: uuid.Generate(),
+			}},
+		}
+		result := aclsAPI.isManagementToken(token)
+		require.True(t, result)
 	})
 }
diff --git a/nomad/consul_test.go b/nomad/consul_test.go
index 79a20a4be32..864d8ba8f76 100644
--- a/nomad/consul_test.go
+++ b/nomad/consul_test.go
@@ -386,9 +386,14 @@ func TestConsulACLsAPI_CheckPermissions(t *testing.T) {
 			try(t, "default", u, "", nil)
 		})
 
-		t.Run("uses kv wrong namespace", func(t *testing.T) {
+		t.Run("uses kv default token missing permissions", func(t *testing.T) {
 			u := &structs.ConsulUsage{KV: true}
-			try(t, "other", u, consul.ExampleOperatorTokenID5, errors.New(`consul ACL token cannot use namespace "other"`))
+			try(t, "other", u, consul.ExampleOperatorTokenID5, errors.New(`insufficient Consul ACL permissions to use template`))
+		})
+
+		t.Run("uses kv token in wrong namespace", func(t *testing.T) {
+			u := &structs.ConsulUsage{KV: true}
+			try(t, "other", u, consul.ExampleOperatorTokenID15, errors.New(`consul ACL token cannot use namespace "other"`))
 		})
 	})
 
@@ -399,8 +404,12 @@ func TestConsulACLsAPI_CheckPermissions(t *testing.T) {
 			try(t, "default", usage, consul.ExampleOperatorTokenID1, nil)
 		})
 
-		t.Run("operator has service wrote wrong ns", func(t *testing.T) {
-			try(t, "other", usage, consul.ExampleOperatorTokenID1, errors.New(`consul ACL token cannot use namespace "other"`))
+		t.Run("operator has service write but no policy", func(t *testing.T) {
+			try(t, "other", usage, consul.ExampleOperatorTokenID1, errors.New(`insufficient Consul ACL permissions to write service "service1"`))
+		})
+
+		t.Run("operator has token in wrong namespace", func(t *testing.T) {
+			try(t, "other", usage, consul.ExampleOperatorTokenID11, errors.New(`consul ACL token cannot use namespace "other"`))
 		})
 
 		t.Run("operator has service_prefix write", func(t *testing.T) {
@@ -434,7 +443,11 @@ func TestConsulACLsAPI_CheckPermissions(t *testing.T) {
 		})
 
 		t.Run("operator has service write wrong ns", func(t *testing.T) {
-			try(t, "other", usage, consul.ExampleOperatorTokenID1, errors.New(`consul ACL token cannot use namespace "other"`))
+			try(t, "other", usage, consul.ExampleOperatorTokenID1, errors.New(`insufficient Consul ACL permissions to write Connect service "service1"`))
+		})
+
+		t.Run("operator has token in wrong namespace", func(t *testing.T) {
+			try(t, "other", usage, consul.ExampleOperatorTokenID11, errors.New(`consul ACL token cannot use namespace "other"`))
 		})
 
 		t.Run("operator has service_prefix write", func(t *testing.T) {