csi: implement volume ACLs (#7339)

* acl/policy: add the volume ACL policies

* nomad/csi_endpoint: enforce ACLs for volume access

* nomad/search_endpoint_oss: volume acls

* acl/acl: add plugin read as a global policy

* acl/policy: add PluginPolicy global cap type

* nomad/csi_endpoint: check the global plugin ACL policy

* nomad/mock/acl: PluginPolicy

* nomad/csi_endpoint: fix list rebase

* nomad/core_sched_test: new test since #7358

* nomad/csi_endpoint_test: use correct permissions for list

* nomad/csi_endpoint: allowCSIMount keeps ACL checks together

* nomad/job_endpoint: check mount permission for jobs

* nomad/job_endpoint_test: need plugin read, too

acl/acl.go

@@ -62,6 +62,7 @@ type ACL struct {
 	node     string
 	operator string
 	quota    string
+	plugin   string
 }
 
 // maxPrivilege returns the policy which grants the most privilege
@@ -193,6 +194,9 @@ func NewACL(management bool, policies []*Policy) (*ACL, error) {
 		if policy.Quota != nil {
 			acl.quota = maxPrivilege(acl.quota, policy.Quota.Policy)
 		}
+		if policy.Plugin != nil {
+			acl.plugin = maxPrivilege(acl.plugin, policy.Plugin.Policy)
+		}
 	}
 
 	// Finalize the namespaces
@@ -477,6 +481,22 @@ func (a *ACL) AllowQuotaWrite() bool {
 	}
 }
 
+// AllowPluginRead checks if read operations are allowed for all plugins
+func (a *ACL) AllowPluginRead() bool {
+	// ACL is nil only if ACLs are disabled
+	if a == nil {
+		return true
+	}
+	switch {
+	case a.management:
+		return true
+	case a.plugin == PolicyRead:
+		return true
+	default:
+		return false
+	}
+}
+
 // IsManagement checks if this represents a management token
 func (a *ACL) IsManagement() bool {
 	return a.management

acl/policy.go

@@ -35,7 +35,10 @@ const (
 	NamespaceCapabilitySentinelOverride = "sentinel-override"
 	NamespaceCapabilityPrivilegedTask   = "privileged-task"
 	NamespaceCapabilityCSIAccess        = "csi-access"
-	NamespaceCapabilityCSICreateVolume  = "csi-create-volume"
+	NamespaceCapabilityCSIWriteVolume   = "csi-write-volume"
+	NamespaceCapabilityCSIReadVolume    = "csi-read-volume"
+	NamespaceCapabilityCSIListVolume    = "csi-list-volume"
+	NamespaceCapabilityCSIMountVolume   = "csi-mount-volume"
 )
 
 var (
@@ -65,6 +68,7 @@ type Policy struct {
 	Node        *NodePolicy         `hcl:"node"`
 	Operator    *OperatorPolicy     `hcl:"operator"`
 	Quota       *QuotaPolicy        `hcl:"quota"`
+	Plugin      *PluginPolicy       `hcl:"plugin"`
 	Raw         string              `hcl:"-"`
 }
 
@@ -76,7 +80,8 @@ func (p *Policy) IsEmpty() bool {
 		p.Agent == nil &&
 		p.Node == nil &&
 		p.Operator == nil &&
-		p.Quota == nil
+		p.Quota == nil &&
+		p.Plugin == nil
 }
 
 // NamespacePolicy is the policy for a specific namespace
@@ -109,6 +114,10 @@ type QuotaPolicy struct {
 	Policy string
 }
 
+type PluginPolicy struct {
+	Policy string
+}
+
 // isPolicyValid makes sure the given string matches one of the valid policies.
 func isPolicyValid(policy string) bool {
 	switch policy {
@@ -126,7 +135,8 @@ func isNamespaceCapabilityValid(cap string) bool {
 		NamespaceCapabilitySubmitJob, NamespaceCapabilityDispatchJob, NamespaceCapabilityReadLogs,
 		NamespaceCapabilityReadFS, NamespaceCapabilityAllocLifecycle,
 		NamespaceCapabilityAllocExec, NamespaceCapabilityAllocNodeExec,
-		NamespaceCapabilityCSIAccess, NamespaceCapabilityCSICreateVolume:
+		NamespaceCapabilityCSIAccess, // TODO(langmartin): remove after plugin caps are done
+		NamespaceCapabilityCSIReadVolume, NamespaceCapabilityCSIWriteVolume, NamespaceCapabilityCSIListVolume, NamespaceCapabilityCSIMountVolume:
 		return true
 	// Separate the enterprise-only capabilities
 	case NamespaceCapabilitySentinelOverride:
@@ -139,25 +149,31 @@ func isNamespaceCapabilityValid(cap string) bool {
 // expandNamespacePolicy provides the equivalent set of capabilities for
 // a namespace policy
 func expandNamespacePolicy(policy string) []string {
+	read := []string{
+		NamespaceCapabilityListJobs,
+		NamespaceCapabilityReadJob,
+		NamespaceCapabilityCSIListVolume,
+		NamespaceCapabilityCSIReadVolume,
+	}
+
+	write := append(read, []string{
+		NamespaceCapabilitySubmitJob,
+		NamespaceCapabilityDispatchJob,
+		NamespaceCapabilityReadLogs,
+		NamespaceCapabilityReadFS,
+		NamespaceCapabilityAllocExec,
+		NamespaceCapabilityAllocLifecycle,
+		NamespaceCapabilityCSIMountVolume,
+		NamespaceCapabilityCSIWriteVolume,
+	}...)
+
 	switch policy {
 	case PolicyDeny:
 		return []string{NamespaceCapabilityDeny}
 	case PolicyRead:
-		return []string{
-			NamespaceCapabilityListJobs,
-			NamespaceCapabilityReadJob,
-		}
+		return read
 	case PolicyWrite:
-		return []string{
-			NamespaceCapabilityListJobs,
-			NamespaceCapabilityReadJob,
-			NamespaceCapabilitySubmitJob,
-			NamespaceCapabilityDispatchJob,
-			NamespaceCapabilityReadLogs,
-			NamespaceCapabilityReadFS,
-			NamespaceCapabilityAllocExec,
-			NamespaceCapabilityAllocLifecycle,
-		}
+		return write
 	default:
 		return nil
 	}
@@ -265,5 +281,9 @@ func Parse(rules string) (*Policy, error) {
 	if p.Quota != nil && !isPolicyValid(p.Quota.Policy) {
 		return nil, fmt.Errorf("Invalid quota policy: %#v", p.Quota)
 	}
+
+	if p.Plugin != nil && !isPolicyValid(p.Plugin.Policy) {
+		return nil, fmt.Errorf("Invalid plugin policy: %#v", p.Plugin)
+	}
 	return p, nil
 }

acl/policy_test.go

@@ -30,6 +30,8 @@ func TestParse(t *testing.T) {
 						Capabilities: []string{
 							NamespaceCapabilityListJobs,
 							NamespaceCapabilityReadJob,
+							NamespaceCapabilityCSIListVolume,
+							NamespaceCapabilityCSIReadVolume,
 						},
 					},
 				},
@@ -58,6 +60,9 @@ func TestParse(t *testing.T) {
 			quota {
 				policy = "read"
 			}
+			plugin {
+				policy = "read"
+			}
 			`,
 			"",
 			&Policy{
@@ -68,6 +73,8 @@ func TestParse(t *testing.T) {
 						Capabilities: []string{
 							NamespaceCapabilityListJobs,
 							NamespaceCapabilityReadJob,
+							NamespaceCapabilityCSIListVolume,
+							NamespaceCapabilityCSIReadVolume,
 						},
 					},
 					{
@@ -76,12 +83,16 @@ func TestParse(t *testing.T) {
 						Capabilities: []string{
 							NamespaceCapabilityListJobs,
 							NamespaceCapabilityReadJob,
+							NamespaceCapabilityCSIListVolume,
+							NamespaceCapabilityCSIReadVolume,
 							NamespaceCapabilitySubmitJob,
 							NamespaceCapabilityDispatchJob,
 							NamespaceCapabilityReadLogs,
 							NamespaceCapabilityReadFS,
 							NamespaceCapabilityAllocExec,
 							NamespaceCapabilityAllocLifecycle,
+							NamespaceCapabilityCSIMountVolume,
+							NamespaceCapabilityCSIWriteVolume,
 						},
 					},
 					{
@@ -104,6 +115,9 @@ func TestParse(t *testing.T) {
 				Quota: &QuotaPolicy{
 					Policy: PolicyRead,
 				},
+				Plugin: &PluginPolicy{
+					Policy: PolicyRead,
+				},
 			},
 		},
 		{