diff --git a/acl/acl.go b/acl/acl.go
index 9a4438ab2..7bd20e2fe 100644
--- a/acl/acl.go
+++ b/acl/acl.go
@@ -61,6 +61,9 @@ type ACL struct {
 	// We use an iradix for the purposes of ordered iteration.
 	wildcardHostVolumes *iradix.Tree
 
+	secureVariables         *iradix.Tree
+	wildcardSecureVariables *iradix.Tree
+
 	agent    string
 	node     string
 	operator string
@@ -98,6 +101,8 @@ func NewACL(management bool, policies []*Policy) (*ACL, error) {
 	wnsTxn := iradix.New().Txn()
 	hvTxn := iradix.New().Txn()
 	whvTxn := iradix.New().Txn()
+	svTxn := iradix.New().Txn()
+	wsvTxn := iradix.New().Txn()
 
 	for _, policy := range policies {
 	NAMESPACES:
@@ -126,6 +131,33 @@ func NewACL(management bool, policies []*Policy) (*ACL, error) {
 				}
 			}
 
+			if ns.SecureVariables != nil {
+				for _, pathPolicy := range ns.SecureVariables.Paths {
+					key := []byte(ns.Name + "\x00" + pathPolicy.PathSpec)
+					var svCapabilities capabilitySet
+					if globDefinition || strings.Contains(pathPolicy.PathSpec, "*") {
+						raw, ok := wsvTxn.Get(key)
+						if ok {
+							svCapabilities = raw.(capabilitySet)
+						} else {
+							svCapabilities = make(capabilitySet)
+						}
+						wsvTxn.Insert(key, svCapabilities)
+					} else {
+						raw, ok := svTxn.Get(key)
+						if ok {
+							svCapabilities = raw.(capabilitySet)
+						} else {
+							svCapabilities = make(capabilitySet)
+						}
+						svTxn.Insert(key, svCapabilities)
+					}
+					for _, cap := range pathPolicy.Capabilities {
+						svCapabilities.Set(cap)
+					}
+				}
+			}
+
 			// Deny always takes precedence
 			if capabilities.Check(NamespaceCapabilityDeny) {
 				continue NAMESPACES
@@ -209,6 +241,8 @@ func NewACL(management bool, policies []*Policy) (*ACL, error) {
 	acl.wildcardNamespaces = wnsTxn.Commit()
 	acl.hostVolumes = hvTxn.Commit()
 	acl.wildcardHostVolumes = whvTxn.Commit()
+	acl.secureVariables = svTxn.Commit()
+	acl.wildcardSecureVariables = wsvTxn.Commit()
 
 	return acl, nil
 }
@@ -324,6 +358,21 @@ func (a *ACL) AllowHostVolume(ns string) bool {
 	return !capabilities.Check(PolicyDeny)
 }
 
+func (a *ACL) AllowSecureVariableOperation(ns, path, op string) bool {
+	if a.management {
+		return true
+	}
+
+	// Check for a matching capability set
+	capabilities, ok := a.matchingSecureVariablesCapabilitySet(ns, path)
+	if !ok {
+		return false
+	}
+
+	return capabilities.Check(op)
+
+}
+
 // matchingNamespaceCapabilitySet looks for a capabilitySet that matches the namespace,
 // if no concrete definitions are found, then we return the closest matching
 // glob.
@@ -392,6 +441,22 @@ func (a *ACL) matchingHostVolumeCapabilitySet(name string) (capabilitySet, bool)
 	return a.findClosestMatchingGlob(a.wildcardHostVolumes, name)
 }
 
+// matchingSecureVariablesCapabilitySet looks for a capabilitySet that matches the namespace and path,
+// if no concrete definitions are found, then we return the closest matching
+// glob.
+// The closest matching glob is the one that has the smallest character
+// difference between the namespace and the glob.
+func (a *ACL) matchingSecureVariablesCapabilitySet(ns, path string) (capabilitySet, bool) {
+	// Check for a concrete matching capability set
+	raw, ok := a.secureVariables.Get([]byte(ns + "\x00" + path))
+	if ok {
+		return raw.(capabilitySet), true
+	}
+
+	// We didn't find a concrete match, so lets try and evaluate globs.
+	return a.findClosestMatchingGlob(a.wildcardSecureVariables, ns+"\x00"+path)
+}
+
 type matchingGlob struct {
 	name          string
 	difference    int
diff --git a/acl/acl_test.go b/acl/acl_test.go
index 872d0728c..c1a5cac15 100644
--- a/acl/acl_test.go
+++ b/acl/acl_test.go
@@ -447,6 +447,115 @@ func TestWildcardHostVolumeMatching(t *testing.T) {
 		})
 	}
 }
+
+func TestSecureVariablesMatching(t *testing.T) {
+	ci.Parallel(t)
+
+	tests := []struct {
+		name   string
+		policy string
+		ns     string
+		path   string
+		op     string
+		allow  bool
+	}{
+		{
+			name: "concrete namespace with concrete path matches",
+			policy: `namespace "ns" {
+					secure_variables { path "foo/bar" { capabilities = ["read"] }}}`,
+			ns:    "ns",
+			path:  "foo/bar",
+			op:    "read",
+			allow: true,
+		},
+		{
+			name: "concrete namespace with concrete path matches for expanded caps",
+			policy: `namespace "ns" {
+					secure_variables { path "foo/bar" { capabilities = ["read"] }}}`,
+			ns:    "ns",
+			path:  "foo/bar",
+			op:    "list",
+			allow: true,
+		},
+		{
+			name: "concrete namespace with wildcard path matches",
+			policy: `namespace "ns" {
+					secure_variables { path "foo/*" { capabilities = ["read"] }}}`,
+			ns:    "ns",
+			path:  "foo/bar",
+			op:    "read",
+			allow: true,
+		},
+		{
+			name: "concrete namespace with invalid concrete path fails",
+			policy: `namespace "ns" {
+					secure_variables { path "bar" { capabilities = ["read"] }}}`,
+			ns:    "ns",
+			path:  "foo/bar",
+			op:    "read",
+			allow: false,
+		},
+		{
+			name: "concrete namespace with invalid wildcard path fails",
+			policy: `namespace "ns" {
+					secure_variables { path "*/foo" { capabilities = ["read"] }}}`,
+			ns:    "ns",
+			path:  "foo/bar",
+			op:    "read",
+			allow: false,
+		},
+		{
+			name: "wildcard namespace with concrete path matches",
+			policy: `namespace "*" {
+					secure_variables { path "foo/bar" { capabilities = ["read"] }}}`,
+			ns:    "ns",
+			path:  "foo/bar",
+			op:    "read",
+			allow: true,
+		},
+		{
+			name: "wildcard namespace with invalid concrete path fails",
+			policy: `namespace "*" {
+					secure_variables { path "bar" { capabilities = ["read"] }}}`,
+			ns:    "ns",
+			path:  "foo/bar",
+			op:    "read",
+			allow: false,
+		},
+		{
+			name: "wildcard in user provided path fails",
+			policy: `namespace "ns" {
+					secure_variables { path "foo/bar" { capabilities = ["read"] }}}`,
+			ns:    "ns",
+			path:  "*",
+			op:    "read",
+			allow: false,
+		},
+		{
+			name: "wildcard attempt to bypass delimiter null byte fails",
+			policy: `namespace "ns" {
+					secure_variables { path "foo/bar" { capabilities = ["read"] }}}`,
+			ns:    "ns*",
+			path:  "bar",
+			op:    "read",
+			allow: false,
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			policy, err := Parse(tc.policy)
+			require.NoError(t, err)
+			require.NotNil(t, policy.Namespaces[0].SecureVariables)
+
+			acl, err := NewACL(false, []*Policy{policy})
+			require.NoError(t, err)
+			require.Equal(t, tc.allow, acl.AllowSecureVariableOperation(tc.ns, tc.path, tc.op))
+		})
+	}
+}
+
 func TestACL_matchingCapabilitySet_returnsAllMatches(t *testing.T) {
 	ci.Parallel(t)
 
diff --git a/acl/policy.go b/acl/policy.go
index 95df4a280..d74df4437 100644
--- a/acl/policy.go
+++ b/acl/policy.go
@@ -67,6 +67,16 @@ var (
 	validVolume = regexp.MustCompile("^[a-zA-Z0-9-*]{1,128}$")
 )
 
+const (
+	// The following are the fine-grained capabilities that can be
+	// granted for a secure variables path. When capabilities are
+	// combined we take the union of all capabilities.
+	SecureVariablesCapabilityList    = "list"
+	SecureVariablesCapabilityRead    = "read"
+	SecureVariablesCapabilityWrite   = "write"
+	SecureVariablesCapabilityDestroy = "destroy"
+)
+
 // Policy represents a parsed HCL or JSON policy.
 type Policy struct {
 	Namespaces  []*NamespacePolicy  `hcl:"namespace,expand"`
@@ -93,8 +103,18 @@ func (p *Policy) IsEmpty() bool {
 
 // NamespacePolicy is the policy for a specific namespace
 type NamespacePolicy struct {
-	Name         string `hcl:",key"`
-	Policy       string
+	Name            string `hcl:",key"`
+	Policy          string
+	Capabilities    []string
+	SecureVariables *SecureVariablesPolicy `hcl:"secure_variables"`
+}
+
+type SecureVariablesPolicy struct {
+	Paths []*SecureVariablesPathPolicy `hcl:"path"`
+}
+
+type SecureVariablesPathPolicy struct {
+	PathSpec     string `hcl:",key"`
 	Capabilities []string
 }
 
@@ -162,6 +182,18 @@ func isNamespaceCapabilityValid(cap string) bool {
 	}
 }
 
+// isPathCapabilityValid ensures the given capability is valid for a
+// secure variables path policy
+func isPathCapabilityValid(cap string) bool {
+	switch cap {
+	case SecureVariablesCapabilityWrite, SecureVariablesCapabilityRead,
+		SecureVariablesCapabilityList, SecureVariablesCapabilityDestroy:
+		return true
+	default:
+		return false
+	}
+}
+
 // expandNamespacePolicy provides the equivalent set of capabilities for
 // a namespace policy
 func expandNamespacePolicy(policy string) []string {
@@ -233,6 +265,22 @@ func expandHostVolumePolicy(policy string) []string {
 	}
 }
 
+func expandSecureVariablesCapabilities(caps []string) []string {
+	var foundRead, foundList bool
+	for _, cap := range caps {
+		switch cap {
+		case SecureVariablesCapabilityRead:
+			foundRead = true
+		case SecureVariablesCapabilityList:
+			foundList = true
+		}
+	}
+	if foundRead && !foundList {
+		caps = append(caps, PolicyList)
+	}
+	return caps
+}
+
 // Parse is used to parse the specified ACL rules into an
 // intermediary set of policies, before being compiled into
 // the ACL
@@ -275,6 +323,24 @@ func Parse(rules string) (*Policy, error) {
 			extraCap := expandNamespacePolicy(ns.Policy)
 			ns.Capabilities = append(ns.Capabilities, extraCap...)
 		}
+
+		if ns.SecureVariables != nil {
+			for _, pathPolicy := range ns.SecureVariables.Paths {
+				if pathPolicy.PathSpec == "" {
+					return nil, fmt.Errorf("Invalid missing secure variable path in namespace %#v", ns)
+				}
+				for _, cap := range pathPolicy.Capabilities {
+					if !isPathCapabilityValid(cap) {
+						return nil, fmt.Errorf(
+							"Invalid secure variable capability '%s' in namespace %#v", cap, ns)
+					}
+				}
+				pathPolicy.Capabilities = expandSecureVariablesCapabilities(pathPolicy.Capabilities)
+
+			}
+
+		}
+
 	}
 
 	for _, hv := range p.HostVolumes {
diff --git a/acl/policy_test.go b/acl/policy_test.go
index e3a8afad6..4ad25fbf9 100644
--- a/acl/policy_test.go
+++ b/acl/policy_test.go
@@ -55,6 +55,19 @@ func TestParse(t *testing.T) {
 			namespace "secret" {
 				capabilities = ["deny", "read-logs"]
 			}
+			namespace "apps" {
+				secure_variables {
+					path "jobs/write-does-not-imply-read-or-delete" {
+						capabilities = ["write"]
+					}
+					path "project/read-implies-list" {
+						capabilities = ["read"]
+					}
+					path "project/explicit" {
+						capabilities = ["read", "list", "destroy"]
+					}
+				}
+			}
 			namespace "autoscaler" {
 				policy = "scale"
 			}
@@ -122,6 +135,32 @@ func TestParse(t *testing.T) {
 							NamespaceCapabilityReadLogs,
 						},
 					},
+					{
+						Name: "apps",
+						SecureVariables: &SecureVariablesPolicy{
+							Paths: []*SecureVariablesPathPolicy{
+								{
+									PathSpec:     "jobs/write-does-not-imply-read-or-delete",
+									Capabilities: []string{SecureVariablesCapabilityWrite},
+								},
+								{
+									PathSpec: "project/read-implies-list",
+									Capabilities: []string{
+										SecureVariablesCapabilityRead,
+										SecureVariablesCapabilityList,
+									},
+								},
+								{
+									PathSpec: "project/explicit",
+									Capabilities: []string{
+										SecureVariablesCapabilityRead,
+										SecureVariablesCapabilityList,
+										SecureVariablesCapabilityDestroy,
+									},
+								},
+							},
+						},
+					},
 					{
 						Name:   "autoscaler",
 						Policy: PolicyScale,
diff --git a/nomad/acl.go b/nomad/acl.go
index d40e37c49..ec26efc4a 100644
--- a/nomad/acl.go
+++ b/nomad/acl.go
@@ -1,6 +1,7 @@
 package nomad
 
 import (
+	"fmt"
 	"time"
 
 	metrics "github.com/armon/go-metrics"
@@ -35,8 +36,50 @@ func (s *Server) ResolveToken(secretID string) (*acl.ACL, error) {
 	return resolveTokenFromSnapshotCache(snap, s.aclCache, secretID)
 }
 
-func (s *Server) ResolveClaim(token string) (*structs.IdentityClaims, error) {
-	return s.encrypter.VerifyClaim(token)
+// VerifyClaim asserts that the token is valid and that the resulting
+// allocation ID belongs to a non-terminal allocation
+func (s *Server) VerifyClaim(token string) (*structs.IdentityClaims, error) {
+
+	claims, err := s.encrypter.VerifyClaim(token)
+	if err != nil {
+		return nil, err
+	}
+	snap, err := s.fsm.State().Snapshot()
+	if err != nil {
+		return nil, err
+	}
+	alloc, err := snap.AllocByID(nil, claims.AllocationID)
+	if err != nil {
+		return nil, err
+	}
+	if alloc == nil || alloc.Job == nil {
+		return nil, fmt.Errorf("allocation does not exist")
+	}
+
+	// the claims for terminal allocs are always treated as expired
+	if alloc.TerminalStatus() {
+		return nil, fmt.Errorf("allocation is terminal")
+	}
+
+	return claims, nil
+}
+
+func (s *Server) ResolveClaims(claims *structs.IdentityClaims) (*acl.ACL, error) {
+
+	policies, err := s.resolvePoliciesForClaims(claims)
+	if err != nil {
+		return nil, err
+	}
+	if len(policies) == 0 {
+		return nil, nil
+	}
+
+	// Compile and cache the ACL object
+	aclObj, err := structs.CompileACLObject(s.aclCache, policies)
+	if err != nil {
+		return nil, err
+	}
+	return aclObj, nil
 }
 
 // resolveTokenFromSnapshotCache is used to resolve an ACL object from a snapshot of state,
@@ -122,3 +165,42 @@ func (s *Server) ResolveSecretToken(secretID string) (*structs.ACLToken, error)
 
 	return token, nil
 }
+
+func (s *Server) resolvePoliciesForClaims(claims *structs.IdentityClaims) ([]*structs.ACLPolicy, error) {
+
+	snap, err := s.fsm.State().Snapshot()
+	if err != nil {
+		return nil, err
+	}
+	alloc, err := snap.AllocByID(nil, claims.AllocationID)
+	if err != nil {
+		return nil, err
+	}
+	if alloc == nil || alloc.Job == nil {
+		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),
+	}
+
+	for _, policyName := range implicitPolicyNames {
+		policy, err := snap.ACLPolicyByName(nil, policyName)
+		if err != nil {
+			return nil, err
+		}
+		if policy == nil {
+			// Ignore policies that don't exist, since they don't
+			// grant any more privilege
+			continue
+		}
+
+		policies = append(policies, policy)
+	}
+	return policies, nil
+}
diff --git a/nomad/secure_variables_endpoint.go b/nomad/secure_variables_endpoint.go
index 37e0674d9..e114c6ea2 100644
--- a/nomad/secure_variables_endpoint.go
+++ b/nomad/secure_variables_endpoint.go
@@ -41,9 +41,11 @@ func (sv *SecureVariables) Upsert(
 	if aclObj, err := sv.srv.ResolveToken(args.AuthToken); err != nil {
 		return err
 	} else if aclObj != nil {
-		// FIXME: Temporary ACL Test policy. Update once implementation complete
-		if !aclObj.AllowNsOp(args.RequestNamespace(), acl.NamespaceCapabilitySubmitJob) {
-			return structs.ErrPermissionDenied
+		for _, variable := range args.Data {
+			if !aclObj.AllowSecureVariableOperation(args.RequestNamespace(),
+				variable.Path, acl.PolicyWrite) {
+				return structs.ErrPermissionDenied
+			}
 		}
 	}
 
@@ -106,8 +108,7 @@ func (sv *SecureVariables) Delete(
 	if aclObj, err := sv.srv.ResolveToken(args.AuthToken); err != nil {
 		return err
 	} else if aclObj != nil {
-		// FIXME: Temporary ACL Test policy. Update once implementation complete
-		if !aclObj.AllowNsOp(args.RequestNamespace(), acl.NamespaceCapabilitySubmitJob) {
+		if !aclObj.AllowSecureVariableOperation(args.RequestNamespace(), args.Path, acl.PolicyWrite) {
 			return structs.ErrPermissionDenied
 		}
 	}
@@ -138,7 +139,7 @@ func (sv *SecureVariables) Read(args *structs.SecureVariablesReadRequest, reply
 
 	// FIXME: Temporary ACL Test policy. Update once implementation complete
 	err := sv.handleMixedAuthEndpoint(args.QueryOptions,
-		acl.NamespaceCapabilitySubmitJob, args.Path)
+		acl.PolicyRead, args.Path)
 	if err != nil {
 		return err
 	}
@@ -190,7 +191,7 @@ func (sv *SecureVariables) List(
 
 	// FIXME: Temporary ACL Test policy. Update once implementation complete
 	err := sv.handleMixedAuthEndpoint(args.QueryOptions,
-		acl.NamespaceCapabilitySubmitJob, args.Prefix)
+		acl.PolicyList, args.Prefix)
 	if err != nil {
 		return err
 	}
@@ -272,8 +273,7 @@ func (s *SecureVariables) listAllSecureVariables(
 	// allowFunc checks whether the caller has the read-job capability on the
 	// passed namespace.
 	allowFunc := func(ns string) bool {
-		// FIXME: Temporary ACL Test policy. Update once implementation complete
-		return aclObj.AllowNsOp(ns, acl.NamespaceCapabilityReadJob)
+		return aclObj.AllowSecureVariableOperation(ns, "", acl.PolicyList)
 	}
 
 	// Set up and return the blocking query.
@@ -390,7 +390,7 @@ func (sv *SecureVariables) handleMixedAuthEndpoint(args structs.QueryOptions, ca
 		// Perform our ACL validation. If the object is nil, this means ACLs
 		// are not enabled, otherwise trigger the allowed namespace function.
 		if aclObj != nil {
-			if !aclObj.AllowNsOp(args.RequestNamespace(), cap) {
+			if !aclObj.AllowSecureVariableOperation(args.RequestNamespace(), pathOrPrefix, cap) {
 				return structs.ErrPermissionDenied
 			}
 		}
@@ -404,12 +404,7 @@ func (sv *SecureVariables) handleMixedAuthEndpoint(args structs.QueryOptions, ca
 
 	// Attempt to verify the token as a JWT with a workload
 	// identity claim
-	claim, err := sv.srv.ResolveClaim(args.AuthToken)
-	if err != nil {
-		return structs.ErrPermissionDenied
-	}
-
-	alloc, err := sv.authValidateAllocationIdentity(claim.AllocationID, args.RequestNamespace())
+	claims, err := sv.srv.VerifyClaim(args.AuthToken)
 	if err != nil {
 		metrics.IncrCounter([]string{
 			"nomad", "secure_variables", "invalid_allocation_identity"}, 1)
@@ -417,49 +412,47 @@ func (sv *SecureVariables) handleMixedAuthEndpoint(args structs.QueryOptions, ca
 		return structs.ErrPermissionDenied
 	}
 
-	err = sv.authValidatePrefix(alloc, claim.TaskName, pathOrPrefix)
+	// The workload identity gets access to paths that match its
+	// identity, without having to go thru the ACL system
+	err = sv.authValidatePrefix(claims, args.RequestNamespace(), pathOrPrefix)
+	if err == nil {
+		return nil
+	}
+
+	// If the workload identity doesn't match the implicit permissions
+	// given to paths, check for its attached ACL policies
+	aclObj, err = sv.srv.ResolveClaims(claims)
 	if err != nil {
-		sv.logger.Trace("allocation identity did not have permission for path", "error", err)
-		return structs.ErrPermissionDenied
+		return err // this only returns an error when the state store has gone wrong
 	}
-
-	return nil
-}
-
-// authValidateAllocationIdentity asserts that the allocation ID
-// belongs to a non-terminal Allocation in the requested namespace
-func (sv *SecureVariables) authValidateAllocationIdentity(allocID, ns string) (*structs.Allocation, error) {
-
-	store, err := sv.srv.fsm.State().Snapshot()
-	if err != nil {
-		return nil, err
+	if aclObj != nil && aclObj.AllowSecureVariableOperation(
+		args.RequestNamespace(), pathOrPrefix, cap) {
+		return nil
 	}
-	alloc, err := store.AllocByID(nil, allocID)
-	if err != nil {
-		return nil, err
-	}
-	if alloc == nil || alloc.Job == nil {
-		return nil, fmt.Errorf("allocation does not exist")
-	}
-
-	// the claims for terminal allocs are always treated as expired
-	if alloc.TerminalStatus() {
-		return nil, fmt.Errorf("allocation is terminal")
-	}
-
-	if alloc.Job.Namespace != ns {
-		return nil, fmt.Errorf("allocation is in another namespace")
-	}
-
-	return alloc, nil
+	return structs.ErrPermissionDenied
 }
 
 // authValidatePrefix asserts that the requested path is valid for
 // this allocation
-func (sv *SecureVariables) authValidatePrefix(alloc *structs.Allocation, taskName, pathOrPrefix string) error {
+func (sv *SecureVariables) authValidatePrefix(claims *structs.IdentityClaims, ns, pathOrPrefix string) error {
+
+	store, err := sv.srv.fsm.State().Snapshot()
+	if err != nil {
+		return err
+	}
+	alloc, err := store.AllocByID(nil, claims.AllocationID)
+	if err != nil {
+		return err
+	}
+	if alloc == nil || alloc.Job == nil {
+		return fmt.Errorf("allocation does not exist")
+	}
+	if alloc.Job.Namespace != ns {
+		return fmt.Errorf("allocation is in another namespace")
+	}
 
 	parts := strings.Split(pathOrPrefix, "/")
-	expect := []string{"jobs", alloc.Job.ID, alloc.TaskGroup, taskName}
+	expect := []string{"jobs", alloc.Job.ID, alloc.TaskGroup, claims.TaskName}
 	if len(parts) > len(expect) {
 		return structs.ErrPermissionDenied
 	}
diff --git a/nomad/secure_variables_endpoint_test.go b/nomad/secure_variables_endpoint_test.go
index 79beb39b1..00ef54bc3 100644
--- a/nomad/secure_variables_endpoint_test.go
+++ b/nomad/secure_variables_endpoint_test.go
@@ -6,6 +6,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/hashicorp/nomad/acl"
 	"github.com/hashicorp/nomad/ci"
 	"github.com/hashicorp/nomad/nomad/mock"
 	"github.com/hashicorp/nomad/nomad/structs"
@@ -24,19 +25,30 @@ func TestSecureVariablesEndpoint_auth(t *testing.T) {
 
 	const ns = "nondefault-namespace"
 
-	alloc := mock.Alloc()
-	alloc.ClientStatus = structs.AllocClientStatusFailed
-	alloc.Job.Namespace = ns
-	jobID := alloc.JobID
+	alloc1 := mock.Alloc()
+	alloc1.ClientStatus = structs.AllocClientStatusFailed
+	alloc1.Job.Namespace = ns
+	alloc1.Namespace = ns
+	jobID := alloc1.JobID
+
+	// create an alloc that will have no access to secure variables we create
+	alloc2 := mock.Alloc()
+	alloc2.Job.TaskGroups[0].Name = "other-no-permissions"
+	alloc2.TaskGroup = "other-no-permissions"
+	alloc2.ClientStatus = structs.AllocClientStatusRunning
+	alloc2.Job.Namespace = ns
+	alloc2.Namespace = ns
 
 	store := srv.fsm.State()
 	require.NoError(t, store.UpsertAllocs(
-		structs.MsgTypeTestSetup, 1000, []*structs.Allocation{alloc}))
+		structs.MsgTypeTestSetup, 1000, []*structs.Allocation{alloc1, alloc2}))
 
-	claim := alloc.ToTaskIdentityClaims("web")
-	e := srv.encrypter
+	claims1 := alloc1.ToTaskIdentityClaims("web")
+	idToken, err := srv.encrypter.SignClaims(claims1)
+	require.NoError(t, err)
 
-	idToken, err := e.SignClaims(claim)
+	claims2 := alloc2.ToTaskIdentityClaims("web")
+	noPermissionsToken, err := srv.encrypter.SignClaims(claims2)
 	require.NoError(t, err)
 
 	// corrupt the signature of the token
@@ -49,6 +61,22 @@ func TestSecureVariablesEndpoint_auth(t *testing.T) {
 	idTokenParts[2] = strings.Join(sig, "")
 	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 "jobs/*" { capabilities = ["read"] }
+		    path "other/path" { capabilities = ["read"] }
+		}}`
+	policy.SetHash()
+	err = store.UpsertACLPolicies(structs.MsgTypeTestSetup, 1100, []*structs.ACLPolicy{policy})
+	require.NoError(t, err)
+
+	aclToken := mock.ACLToken()
+	aclToken.Policies = []string{policy.Name}
+	err = store.UpsertACLTokens(structs.MsgTypeTestSetup, 1150, []*structs.ACLToken{aclToken})
+	require.NoError(t, err)
+
 	t.Run("terminal alloc should be denied", func(t *testing.T) {
 		err = srv.staticEndpoints.SecureVariables.handleMixedAuthEndpoint(
 			structs.QueryOptions{AuthToken: idToken, Namespace: ns}, "n/a",
@@ -57,9 +85,9 @@ func TestSecureVariablesEndpoint_auth(t *testing.T) {
 	})
 
 	// make alloc non-terminal
-	alloc.ClientStatus = structs.AllocClientStatusRunning
+	alloc1.ClientStatus = structs.AllocClientStatusRunning
 	require.NoError(t, store.UpsertAllocs(
-		structs.MsgTypeTestSetup, 1200, []*structs.Allocation{alloc}))
+		structs.MsgTypeTestSetup, 1200, []*structs.Allocation{alloc1}))
 
 	t.Run("wrong namespace should be denied", func(t *testing.T) {
 		err = srv.staticEndpoints.SecureVariables.handleMixedAuthEndpoint(
@@ -103,6 +131,48 @@ func TestSecureVariablesEndpoint_auth(t *testing.T) {
 			path:        "jobs",
 			expectedErr: nil,
 		},
+		{
+			name:        "valid claim for implied policy",
+			token:       idToken,
+			cap:         acl.PolicyRead,
+			path:        "other/path",
+			expectedErr: nil,
+		},
+		{
+			name:        "valid claim for implied policy path denied",
+			token:       idToken,
+			cap:         acl.PolicyRead,
+			path:        "other/not-allowed",
+			expectedErr: structs.ErrPermissionDenied,
+		},
+		{
+			name:        "valid claim for implied policy capability denied",
+			token:       idToken,
+			cap:         acl.PolicyWrite,
+			path:        "other/path",
+			expectedErr: structs.ErrPermissionDenied,
+		},
+		{
+			name:        "valid claim with no permissions denied by path",
+			token:       noPermissionsToken,
+			cap:         "n/a",
+			path:        fmt.Sprintf("jobs/%s/w", jobID),
+			expectedErr: structs.ErrPermissionDenied,
+		},
+		{
+			name:        "valid claim with no permissions allowed by namespace",
+			token:       noPermissionsToken,
+			cap:         "n/a",
+			path:        "jobs",
+			expectedErr: nil,
+		},
+		{
+			name:        "valid claim with no permissions denied by capability",
+			token:       noPermissionsToken,
+			cap:         acl.PolicyRead,
+			path:        fmt.Sprintf("jobs/%s/w", jobID),
+			expectedErr: structs.ErrPermissionDenied,
+		},
 		{
 			name:        "extra trailing slash is denied",
 			token:       idToken,
@@ -130,6 +200,20 @@ func TestSecureVariablesEndpoint_auth(t *testing.T) {
 			path:        fmt.Sprintf("jobs/%s/web/web", jobID),
 			expectedErr: structs.ErrPermissionDenied,
 		},
+		{
+			name:        "acl token read policy is allowed to list",
+			token:       aclToken.SecretID,
+			cap:         acl.PolicyList,
+			path:        fmt.Sprintf("jobs/%s/web/web", jobID),
+			expectedErr: nil,
+		},
+		{
+			name:        "acl token read policy is not allowed to write",
+			token:       aclToken.SecretID,
+			cap:         acl.PolicyWrite,
+			path:        fmt.Sprintf("jobs/%s/web/web", jobID),
+			expectedErr: structs.ErrPermissionDenied,
+		},
 	}
 
 	for _, tc := range testCases {
diff --git a/nomad/service_registration_endpoint.go b/nomad/service_registration_endpoint.go
index 9428707df..93406fec5 100644
--- a/nomad/service_registration_endpoint.go
+++ b/nomad/service_registration_endpoint.go
@@ -533,7 +533,7 @@ func (s *ServiceRegistration) handleMixedAuthEndpoint(args structs.QueryOptions,
 		// identity claim if it's not a secret ID.
 		// COMPAT(1.4.0): we can remove this conditional in 1.5.0
 		if !helper.IsUUID(args.AuthToken) {
-			claims, err := s.srv.ResolveClaim(args.AuthToken)
+			claims, err := s.srv.VerifyClaim(args.AuthToken)
 			if err != nil {
 				return err
 			}