diff --git a/api/acl.go b/api/acl.go
index 486bbcb5e3a0..435b5c3c4812 100644
--- a/api/acl.go
+++ b/api/acl.go
@@ -212,11 +212,15 @@ type ACLPolicyListStub struct {
 
 // ACLPolicy is used to represent an ACL policy
 type ACLPolicy struct {
-	Name        string
-	Description string
-	Rules       string
-	CreateIndex uint64
-	ModifyIndex uint64
+	Name         string
+	Description  string
+	Rules        string
+	JobNamespace string
+	JobID        string
+	Group        string
+	Task         string
+	CreateIndex  uint64
+	ModifyIndex  uint64
 }
 
 // ACLToken represents a client token which is used to Authenticate
diff --git a/command/acl_bootstrap.go b/command/acl_bootstrap.go
index f8970f938499..87c77b0a2454 100644
--- a/command/acl_bootstrap.go
+++ b/command/acl_bootstrap.go
@@ -130,16 +130,33 @@ func (c *ACLBootstrapCommand) Run(args []string) int {
 	return 0
 }
 
-// formatKVPolicy returns a K/V formatted policy
-func formatKVPolicy(policy *api.ACLPolicy) string {
+// formatACLPolicy returns formatted policy
+func formatACLPolicy(policy *api.ACLPolicy) string {
 	output := []string{
 		fmt.Sprintf("Name|%s", policy.Name),
 		fmt.Sprintf("Description|%s", policy.Description),
-		fmt.Sprintf("Rules|%s", policy.Rules),
 		fmt.Sprintf("CreateIndex|%v", policy.CreateIndex),
 		fmt.Sprintf("ModifyIndex|%v", policy.ModifyIndex),
 	}
-	return formatKV(output)
+
+	formattedOut := formatKV(output)
+
+	if policy.JobNamespace != "" {
+		output := []string{
+			fmt.Sprintf("Namespace|%v", policy.JobNamespace),
+			fmt.Sprintf("JobID|%v", policy.JobID),
+			fmt.Sprintf("Group|%v", policy.Group),
+			fmt.Sprintf("Task|%v", policy.Task),
+		}
+		formattedOut += "\n\n[bold]Associated Workload[reset]\n"
+		formattedOut += formatKV(output)
+	}
+
+	// these are potentially large blobs so leave till the end
+	formattedOut += "\n\n[bold]Rules[reset]\n\n"
+	formattedOut += policy.Rules
+
+	return formattedOut
 }
 
 // formatKVACLToken returns a K/V formatted ACL token
diff --git a/command/acl_policy_apply.go b/command/acl_policy_apply.go
index 02c5ef45618d..64814c8b0fa4 100644
--- a/command/acl_policy_apply.go
+++ b/command/acl_policy_apply.go
@@ -32,6 +32,21 @@ Apply Options:
   -description
     Specifies a human readable description for the policy.
 
+  -job
+    Attaches the policy to the specified job. Requires that -namespace is
+    also set.
+
+  -namespace
+    Attaches the policy to the specified namespace. Requires that -job is
+    also set.
+
+  -group
+    Attaches the policy to the specified task group. Requires that -namespace
+    and -job are also set.
+
+  -task
+    Attaches the policy to the specified task. Requires that -namespace, -job
+    and -group are also set.
 `
 	return strings.TrimSpace(helpText)
 }
@@ -53,9 +68,16 @@ func (c *ACLPolicyApplyCommand) Name() string { return "acl policy apply" }
 
 func (c *ACLPolicyApplyCommand) Run(args []string) int {
 	var description string
+	var jobID, group, task string // namespace is included in default flagset
+
 	flags := c.Meta.FlagSet(c.Name(), FlagSetClient)
 	flags.Usage = func() { c.Ui.Output(c.Help()) }
 	flags.StringVar(&description, "description", "", "")
+
+	flags.StringVar(&jobID, "job", "", "job to attach this policy to")
+	flags.StringVar(&group, "group", "", "group to attach this policy to")
+	flags.StringVar(&task, "task", "", "task to attach this policy to")
+
 	if err := flags.Parse(args); err != nil {
 		return 1
 	}
@@ -89,11 +111,31 @@ func (c *ACLPolicyApplyCommand) Run(args []string) int {
 		}
 	}
 
+	f := flags.Lookup("namespace")
+	namespace := f.Value.String()
+
+	if namespace == "" && jobID != "" {
+		c.Ui.Error("-namespace is required if -job is set")
+		return 1
+	}
+	if jobID == "" && group != "" {
+		c.Ui.Error("-job is required if -group is set")
+		return 1
+	}
+	if group == "" && task != "" {
+		c.Ui.Error("-group is required if -task is set")
+		return 1
+	}
+
 	// Construct the policy
 	ap := &api.ACLPolicy{
-		Name:        policyName,
-		Description: description,
-		Rules:       string(rawPolicy),
+		Name:         policyName,
+		Description:  description,
+		Rules:        string(rawPolicy),
+		JobNamespace: namespace,
+		JobID:        jobID,
+		Group:        group,
+		Task:         task,
 	}
 
 	// Get the HTTP client
diff --git a/command/acl_policy_info.go b/command/acl_policy_info.go
index beeae5f293fc..4ddf320e58da 100644
--- a/command/acl_policy_info.go
+++ b/command/acl_policy_info.go
@@ -74,6 +74,6 @@ func (c *ACLPolicyInfoCommand) Run(args []string) int {
 		return 1
 	}
 
-	c.Ui.Output(formatKVPolicy(policy))
+	c.Ui.Output(c.Colorize().Color(formatACLPolicy(policy)))
 	return 0
 }
diff --git a/nomad/acl.go b/nomad/acl.go
index ec26efc4a6ba..59872bd36d89 100644
--- a/nomad/acl.go
+++ b/nomad/acl.go
@@ -180,27 +180,30 @@ func (s *Server) resolvePoliciesForClaims(claims *structs.IdentityClaims) ([]*st
 		return nil, fmt.Errorf("allocation does not exist")
 	}
 
-	// Find any implicit policies associated with this task
-	policies := []*structs.ACLPolicy{}
-	implicitPolicyNames := []string{
-		fmt.Sprintf("_:%s/%s/%s/%s", alloc.Namespace, alloc.Job.ID, alloc.TaskGroup, claims.TaskName),
-		fmt.Sprintf("_:%s/%s/%s", alloc.Namespace, alloc.Job.ID, alloc.TaskGroup),
-		fmt.Sprintf("_:%s/%s", alloc.Namespace, alloc.Job.ID),
-		fmt.Sprintf("_:%s", alloc.Namespace),
+	// Find any policies attached to the job
+	iter, err := snap.ACLPolicyByJob(nil, alloc.Namespace, alloc.Job.ID)
+	if err != nil {
+		return nil, err
 	}
-
-	for _, policyName := range implicitPolicyNames {
-		policy, err := snap.ACLPolicyByName(nil, policyName)
-		if err != nil {
-			return nil, err
+	policies := []*structs.ACLPolicy{}
+	for {
+		raw := iter.Next()
+		if raw == nil {
+			break
 		}
-		if policy == nil {
-			// Ignore policies that don't exist, since they don't
-			// grant any more privilege
-			continue
+		policy := raw.(*structs.ACLPolicy)
+
+		switch {
+		case policy.Group == "":
+			policies = append(policies, policy)
+		case policy.Group != alloc.TaskGroup:
+			continue // don't bother checking task
+		case policy.Task == "":
+			policies = append(policies, policy)
+		case policy.Task == claims.TaskName:
+			policies = append(policies, policy)
 		}
-
-		policies = append(policies, policy)
 	}
+
 	return policies, nil
 }
diff --git a/nomad/acl_test.go b/nomad/acl_test.go
index 867150639aa2..45c4e0dbe739 100644
--- a/nomad/acl_test.go
+++ b/nomad/acl_test.go
@@ -4,6 +4,9 @@ import (
 	"testing"
 
 	lru "github.com/hashicorp/golang-lru"
+	"github.com/shoenig/test/must"
+	"github.com/stretchr/testify/assert"
+
 	"github.com/hashicorp/nomad/acl"
 	"github.com/hashicorp/nomad/ci"
 	"github.com/hashicorp/nomad/helper/uuid"
@@ -11,7 +14,6 @@ import (
 	"github.com/hashicorp/nomad/nomad/state"
 	"github.com/hashicorp/nomad/nomad/structs"
 	"github.com/hashicorp/nomad/testutil"
-	"github.com/stretchr/testify/assert"
 )
 
 func TestResolveACLToken(t *testing.T) {
@@ -63,7 +65,7 @@ func TestResolveACLToken(t *testing.T) {
 	assert.Nil(t, err)
 	assert.NotNil(t, aclObj)
 
-	// Check that the ACL object is sane
+	// Check that the ACL object looks reasonable
 	assert.Equal(t, false, aclObj.IsManagement())
 	allowed := aclObj.AllowNamespaceOperation("default", acl.NamespaceCapabilityListJobs)
 	assert.Equal(t, true, allowed)
@@ -132,3 +134,102 @@ func TestResolveSecretToken(t *testing.T) {
 	}
 
 }
+
+func TestResolveClaims(t *testing.T) {
+	ci.Parallel(t)
+
+	srv, _, cleanup := TestACLServer(t, nil)
+	defer cleanup()
+
+	store := srv.fsm.State()
+	index := uint64(100)
+
+	alloc := mock.Alloc()
+
+	claims := &structs.IdentityClaims{
+		Namespace:    alloc.Namespace,
+		JobID:        alloc.Job.ID,
+		AllocationID: alloc.ID,
+		TaskName:     alloc.Job.TaskGroups[0].Tasks[0].Name,
+	}
+
+	// unrelated policy
+	policy0 := mock.ACLPolicy()
+
+	// policy for job
+	policy1 := mock.ACLPolicy()
+	policy1.JobNamespace = claims.Namespace
+	policy1.JobID = claims.JobID
+
+	// policy for job and group
+	policy2 := mock.ACLPolicy()
+	policy2.JobNamespace = claims.Namespace
+	policy2.JobID = claims.JobID
+	policy2.Group = alloc.Job.TaskGroups[0].Name
+
+	// policy for job and group	and task
+	policy3 := mock.ACLPolicy()
+	policy3.JobNamespace = claims.Namespace
+	policy3.JobID = claims.JobID
+	policy3.Group = alloc.Job.TaskGroups[0].Name
+	policy3.Task = claims.TaskName
+
+	// policy for job and group	but different task
+	policy4 := mock.ACLPolicy()
+	policy4.JobNamespace = claims.Namespace
+	policy4.JobID = claims.JobID
+	policy4.Group = alloc.Job.TaskGroups[0].Name
+	policy4.Task = "another"
+
+	// policy for job but different group
+	policy5 := mock.ACLPolicy()
+	policy5.JobNamespace = claims.Namespace
+	policy5.JobID = claims.JobID
+	policy5.Group = "another"
+
+	// policy for same namespace but different job
+	policy6 := mock.ACLPolicy()
+	policy6.JobNamespace = claims.Namespace
+	policy6.JobID = "another"
+
+	// policy for same job in different namespace
+	policy7 := mock.ACLPolicy()
+	policy7.JobNamespace = "another"
+	policy7.JobID = claims.JobID
+
+	index++
+	err := store.UpsertACLPolicies(structs.MsgTypeTestSetup, index, []*structs.ACLPolicy{
+		policy0, policy1, policy2, policy3, policy4, policy5, policy6, policy7})
+	must.NoError(t, err)
+
+	aclObj, err := srv.ResolveClaims(claims)
+	must.Nil(t, aclObj)
+	must.EqError(t, err, "allocation does not exist")
+
+	// upsert the allocation
+	index++
+	err = store.UpsertAllocs(structs.MsgTypeTestSetup, index, []*structs.Allocation{alloc})
+	must.NoError(t, err)
+
+	aclObj, err = srv.ResolveClaims(claims)
+	must.NoError(t, err)
+	must.NotNil(t, aclObj)
+
+	// Check that the ACL object looks reasonable
+	must.False(t, aclObj.IsManagement())
+	must.True(t, aclObj.AllowNamespaceOperation("default", acl.NamespaceCapabilityListJobs))
+	must.False(t, aclObj.AllowNamespaceOperation("other", acl.NamespaceCapabilityListJobs))
+
+	// Resolve the same claim again, should get cache value
+	aclObj2, err := srv.ResolveClaims(claims)
+	must.NoError(t, err)
+	must.NotNil(t, aclObj)
+	must.Eq(t, aclObj, aclObj2, must.Sprintf("expected cached value"))
+
+	policies, err := srv.resolvePoliciesForClaims(claims)
+	must.NoError(t, err)
+	must.Len(t, 3, policies)
+	must.Contains(t, policies, policy1)
+	must.Contains(t, policies, policy2)
+	must.Contains(t, policies, policy3)
+}
diff --git a/nomad/secure_variables_endpoint_test.go b/nomad/secure_variables_endpoint_test.go
index 58aafd892080..7d373e21f09d 100644
--- a/nomad/secure_variables_endpoint_test.go
+++ b/nomad/secure_variables_endpoint_test.go
@@ -76,12 +76,14 @@ func TestSecureVariablesEndpoint_auth(t *testing.T) {
 	invalidIDToken := strings.Join(idTokenParts, ".")
 
 	policy := mock.ACLPolicy()
-	policy.Name = fmt.Sprintf("_:%s/%s/%s", ns, jobID, alloc1.TaskGroup)
 	policy.Rules = `namespace "nondefault-namespace" {
 		secure_variables {
 		    path "nomad/jobs/*" { capabilities = ["read"] }
 		    path "other/path" { capabilities = ["read"] }
 		}}`
+	policy.JobNamespace = ns
+	policy.JobID = jobID
+	policy.Group = alloc1.TaskGroup
 	policy.SetHash()
 	err = store.UpsertACLPolicies(structs.MsgTypeTestSetup, 1100, []*structs.ACLPolicy{policy})
 	must.NoError(t, err)
diff --git a/nomad/state/schema.go b/nomad/state/schema.go
index 13e0a3264bf2..6dcb15f17f70 100644
--- a/nomad/state/schema.go
+++ b/nomad/state/schema.go
@@ -775,6 +775,21 @@ func aclPolicyTableSchema() *memdb.TableSchema {
 					Field: "Name",
 				},
 			},
+			"job": {
+				Name:         "job",
+				AllowMissing: true,
+				Unique:       false,
+				Indexer: &memdb.CompoundIndex{
+					Indexes: []memdb.Indexer{
+						&memdb.StringFieldIndex{
+							Field: "JobNamespace",
+						},
+						&memdb.StringFieldIndex{
+							Field: "JobID",
+						},
+					},
+				},
+			},
 		},
 	}
 }
diff --git a/nomad/state/state_store.go b/nomad/state/state_store.go
index 737dd6b22a0f..0152ccc80b3f 100644
--- a/nomad/state/state_store.go
+++ b/nomad/state/state_store.go
@@ -5570,6 +5570,20 @@ func (s *StateStore) ACLPolicyByNamePrefix(ws memdb.WatchSet, prefix string) (me
 	return iter, nil
 }
 
+// ACLPolicyByJob is used to lookup policies that have been attached to a
+// specific job
+func (s *StateStore) ACLPolicyByJob(ws memdb.WatchSet, ns, jobID string) (memdb.ResultIterator, error) {
+	txn := s.db.ReadTxn()
+
+	iter, err := txn.Get("acl_policy", "job_prefix", ns, jobID)
+	if err != nil {
+		return nil, fmt.Errorf("acl policy lookup failed: %v", err)
+	}
+	ws.Add(iter.WatchCh())
+
+	return iter, nil
+}
+
 // ACLPolicies returns an iterator over all the acl policies
 func (s *StateStore) ACLPolicies(ws memdb.WatchSet) (memdb.ResultIterator, error) {
 	txn := s.db.ReadTxn()
diff --git a/nomad/structs/structs.go b/nomad/structs/structs.go
index 824eddc0958a..28388d9e479c 100644
--- a/nomad/structs/structs.go
+++ b/nomad/structs/structs.go
@@ -11792,6 +11792,12 @@ type ACLPolicy struct {
 	Rules       string      // HCL or JSON format
 	RulesJSON   *acl.Policy // Generated from Rules on read
 	Hash        []byte
+
+	JobNamespace string // namespace of the job this policy is attached to
+	JobID        string // ID of the job this policy is attached to
+	Group        string // ID of the group this policy is attached to
+	Task         string // ID of the task this policy is attached to
+
 	CreateIndex uint64
 	ModifyIndex uint64
 }
@@ -11808,6 +11814,10 @@ func (a *ACLPolicy) SetHash() []byte {
 	_, _ = hash.Write([]byte(a.Name))
 	_, _ = hash.Write([]byte(a.Description))
 	_, _ = hash.Write([]byte(a.Rules))
+	_, _ = hash.Write([]byte(a.JobNamespace))
+	_, _ = hash.Write([]byte(a.JobID))
+	_, _ = hash.Write([]byte(a.Group))
+	_, _ = hash.Write([]byte(a.Task))
 
 	// Finalize the hash
 	hashVal := hash.Sum(nil)
@@ -11841,6 +11851,19 @@ func (a *ACLPolicy) Validate() error {
 		err := fmt.Errorf("description longer than %d", maxPolicyDescriptionLength)
 		mErr.Errors = append(mErr.Errors, err)
 	}
+	if a.JobNamespace == "" && a.JobID != "" {
+		err := fmt.Errorf("namespace must be set to set job ID")
+		mErr.Errors = append(mErr.Errors, err)
+	}
+	if a.JobID == "" && a.Group != "" {
+		err := fmt.Errorf("job ID must be set to set group")
+		mErr.Errors = append(mErr.Errors, err)
+	}
+	if a.Group == "" && a.Task != "" {
+		err := fmt.Errorf("group must be set to set task")
+		mErr.Errors = append(mErr.Errors, err)
+	}
+
 	return mErr.ErrorOrNil()
 }
 
diff --git a/website/content/docs/commands/acl/policy-apply.mdx b/website/content/docs/commands/acl/policy-apply.mdx
index 3a01ca5cb24a..af000cd896aa 100644
--- a/website/content/docs/commands/acl/policy-apply.mdx
+++ b/website/content/docs/commands/acl/policy-apply.mdx
@@ -28,6 +28,19 @@ This command requires a management ACL token.
 
 - `-description`: Sets the human readable description for the ACL policy.
 
+- `-job`: Attaches the policy to the specified job. Requires that `-namespace` is
+    also set.
+
+- `-namespace`: Attaches the policy to the specified namespace. Requires that
+    `-job` is also set.
+
+- `-group`: Attaches the policy to the specified task group. Requires that
+    `-namespace` and `-job` are also set.
+
+- `-task`: Attaches the policy to the specified task. Requires that `-namespace`,
+    `-job` and `-group` are also set.
+
+
 ## Examples
 
 Create a new ACL Policy:
@@ -36,3 +49,12 @@ Create a new ACL Policy:
 $ nomad acl policy apply my-policy my-policy.json
 Successfully wrote 'my-policy' ACL policy!
 ```
+
+Associate an ACL Policy with a specific task:
+
+```shell-session
+$ nomad acl policy apply \
+    -namespace default -job example -group cache -task redis \
+    my-policy my-policy.json
+Successfully wrote 'my-policy' ACL policy!
+```
diff --git a/website/content/docs/commands/acl/policy-info.mdx b/website/content/docs/commands/acl/policy-info.mdx
index 317a91bde97e..b9e9e99a4b10 100644
--- a/website/content/docs/commands/acl/policy-info.mdx
+++ b/website/content/docs/commands/acl/policy-info.mdx
@@ -34,11 +34,38 @@ Fetch information on an existing ACL Policy:
 $ nomad acl policy info my-policy
 Name        = my-policy
 Description = <none>
-Rules       = {
+CreateIndex = 749
+ModifyIndex = 758
+
+Rules
+
+{
   "Name": "my-policy",
   "Description": "This is a great policy",
   "Rules": "list_jobs"
 }
+```
+
+If the ACL Policy is associated with a [Workload Identity], additional information will be shown:
+
+```shell-session
+$ nomad acl policy info my-policy
+Name        = my-policy
+Description = <none>
 CreateIndex = 749
 ModifyIndex = 758
+
+Associated Workload
+Namespace = default
+JobID     = example
+Group     = cache
+Task      = redis
+
+Rules
+
+{
+  "Name": "my-policy",
+  "Description": "This is a great policy",
+  "Rules": "list_jobs"
+}
 ```
diff --git a/website/content/docs/concepts/secure-variables.mdx b/website/content/docs/concepts/secure-variables.mdx
index eb5254d189ec..944cdeb812b4 100644
--- a/website/content/docs/concepts/secure-variables.mdx
+++ b/website/content/docs/concepts/secure-variables.mdx
@@ -131,8 +131,9 @@ namespace "default" {
 ```
 
 You can provide access to additional secrets by creating policies associated
-with the task's [workload identity]. For example, to give the task above access to
-set of shared secrets, you can create the following policy file:
+with the task's [workload identity]. For example, to give the task above access
+to all secrets in the "shared" namespace, you can create the following policy
+file:
 
 ```hcl
 namespace "shared" {
@@ -144,11 +145,12 @@ namespace "shared" {
 }
 ```
 
-Then create the policy with to give the task read access to
-all paths in the "shared" namespace:
+Then create the policy and associate it with the specific task:
 
 ```shell-session
-nomad acl policy apply "_:/default/example/cache/redis" ./policy.hcl
+nomad acl policy apply \
+   -namespace default -job example -group cache -task redis \
+   redis-policy ./policy.hcl
 ```
 
 See [Implicit Access to ACL Policies] for more details.
diff --git a/website/content/docs/concepts/workload-identity.mdx b/website/content/docs/concepts/workload-identity.mdx
index a71472d7a6ce..afb59382d067 100644
--- a/website/content/docs/concepts/workload-identity.mdx
+++ b/website/content/docs/concepts/workload-identity.mdx
@@ -20,33 +20,16 @@ workload identity includes the following identity claims:
 }
 ```
 
-## Implicit Access to ACL Policies
+# Workload Associated ACL Policies
 
-Nomad automatically attaches a set of implicit ACL policies to every workload
-identity. The names of these policies start with the Nomad-owned prefix `_:`,
-followed by the namespace, job ID, task group name, and task name.
+You can associate additional ACL policies with workload identities by passing
+the `-job`, `-group`, and `-task` flags to `nomad acl policy apply`. When Nomad
+resolves a workload identity claim, it will automatically include policies that
+match. If no matching policies exist, the workload identity does not have any
+additional capabilities.
 
-```
-_:/$namespace/$job_id/$task_group/$task
-_:/$namespace/$job_id/$task_group
-_:/$namespace/$job_id
-_:/$namespace
-```
-
-For example, a task named "redis", in a group named "cache", in a job named
-"example", will automatically have the following policies:
-
-```
-_:/default/example/cache/redis
-_:/default/example/cache
-_:/default/example
-_:/default
-```
-
-If these policies do not exist, the workload identity does not have any
-additional capabilities. But you can create a policy with one of these names and
-the task will automatically have access to them. For example, to give the task
-above access to set of shared secrets, you can create the following policy file:
+For example, to allow a workload access to secrets from the namespace "shared",
+you can create the following policy file:
 
 ```hcl
 namespace "shared" {
@@ -58,11 +41,30 @@ namespace "shared" {
 }
 ```
 
-Then create the policy to give the task read access to
-all paths in the "shared" namespace:
+You can then apply this policy to a specific task:
+
+```shell-session
+nomad acl policy apply \
+   -namespace default -job example -group cache -task redis \
+   redis-policy ./policy.hcl
+```
+
+You can also apply this policy to all tasks in the group by omitting the `-task`
+flag:
+
+```shell-session
+nomad acl policy apply \
+   -namespace default -job example -group cache \
+   redis-policy ./policy.hcl
+```
+
+And you can apply this policy to all groups in the job by omitting both the
+`-group` and `-task` flag:
 
 ```shell-session
-nomad acl policy apply "_:/default/example/cache/redis" ./policy.hcl
+nomad acl policy apply \
+   -namespace default -job example \
+   redis-policy ./policy.hcl
 ```
 
 ## Using Workload Identity