job endpoint hooks to enforce access to vault/consul clusters (CE)

nomad/job_endpoint.go

@@ -73,12 +73,15 @@ func NewJobEndpoints(s *Server, ctx *RPCContext) *Job {
 			jobExposeCheckHook{},
 			jobImpliedConstraints{},
 			jobNodePoolMutatingHook{srv: s},
+			jobVaultHook{srv: s},
+			jobConsulHook{srv: s},
 			jobImplicitIdentitiesHook{srv: s},
 		},
 		validators: []jobValidator{
 			jobConnectHook{},
 			jobExposeCheckHook{},
 			jobVaultHook{srv: s},
+			jobConsulHook{srv: s},
 			jobNamespaceConstraintCheckHook{srv: s},
 			jobNodePoolValidatingHook{srv: s},
 			&jobValidate{srv: s},

nomad/job_endpoint_hook_connect_test.go

@@ -11,9 +11,9 @@ import (
 	"github.com/hashicorp/nomad/ci"
 	"github.com/hashicorp/nomad/helper"
 	"github.com/hashicorp/nomad/helper/pointer"
-	"github.com/hashicorp/nomad/helper/testlog"
 	"github.com/hashicorp/nomad/nomad/mock"
 	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/hashicorp/nomad/testutil"
 	"github.com/shoenig/test/must"
 	"github.com/stretchr/testify/require"
 )
@@ -407,17 +407,22 @@ func TestJobEndpointConnect_groupConnectHook_MeshGateway(t *testing.T) {
 func TestJobEndpointConnect_ConnectInterpolation(t *testing.T) {
 	ci.Parallel(t)
 
-	server := &Server{logger: testlog.HCLogger(t), config: DefaultConfig()}
-	jobEndpoint := NewJobEndpoints(server, nil)
+	srv, cleanup := TestServer(t, func(c *Config) {
+		c.NumSchedulers = 0
+	})
+	t.Cleanup(cleanup)
+	testutil.WaitForLeader(t, srv.RPC)
+
+	jobEndpoint := NewJobEndpoints(srv, nil)
 
 	j := mock.ConnectJob()
 	j.TaskGroups[0].Services[0].Name = "${JOB}-api"
 	j, warnings, err := jobEndpoint.admissionMutators(j)
-	require.NoError(t, err)
-	require.Nil(t, warnings)
+	must.NoError(t, err)
+	must.Nil(t, warnings)
 
-	require.Len(t, j.TaskGroups[0].Tasks, 2)
-	require.Equal(t, "connect-proxy-my-job-api", j.TaskGroups[0].Tasks[1].Name)
+	must.Len(t, 2, j.TaskGroups[0].Tasks)
+	must.Eq(t, "connect-proxy-my-job-api", j.TaskGroups[0].Tasks[1].Name)
 }
 
 func TestJobEndpointConnect_groupConnectSidecarValidate(t *testing.T) {

nomad/job_endpoint_hook_consul.go

@@ -0,0 +1,14 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package nomad
+
+// jobConsulHook is a job registration admission controller for Consul
+// configuration in Consul, Service, and Template blocks
+type jobConsulHook struct {
+	srv *Server
+}
+
+func (jobConsulHook) Name() string {
+	return "consul"
+}

nomad/job_endpoint_hook_consul_ce.go

@@ -0,0 +1,77 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+//go:build !ent
+// +build !ent
+
+package nomad
+
+import (
+	"errors"
+
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+func (h jobConsulHook) Validate(job *structs.Job) ([]error, error) {
+
+	for _, group := range job.TaskGroups {
+		if group.Consul != nil {
+			if err := h.validateCluster(group.Consul.Cluster); err != nil {
+				return nil, err
+			}
+		}
+
+		for _, service := range group.Services {
+			if service.Provider == structs.ServiceProviderConsul {
+				if err := h.validateCluster(service.Cluster); err != nil {
+					return nil, err
+				}
+			}
+		}
+
+		for _, task := range group.Tasks {
+			for _, service := range task.Services {
+				if service.Provider == structs.ServiceProviderConsul {
+					if err := h.validateCluster(service.Cluster); err != nil {
+						return nil, err
+					}
+				}
+			}
+		}
+	}
+
+	return nil, nil
+}
+
+func (h jobConsulHook) validateCluster(name string) error {
+	if name != "default" {
+		return errors.New("non-default Consul cluster requires Nomad Enterprise")
+	}
+	return nil
+}
+
+// Mutate ensures that the job's Consul cluster has been configured to be the
+// default Consul cluster if unset
+func (j jobConsulHook) Mutate(job *structs.Job) (*structs.Job, []error, error) {
+	for _, group := range job.TaskGroups {
+		if group.Consul != nil && group.Consul.Cluster == "" {
+			group.Consul.Cluster = "default"
+		}
+
+		for _, service := range group.Services {
+			if service.Provider == structs.ServiceProviderConsul && service.Cluster == "" {
+				service.Cluster = "default"
+			}
+		}
+
+		for _, task := range group.Tasks {
+			for _, service := range task.Services {
+				if service.Provider == structs.ServiceProviderConsul && service.Cluster == "" {
+					service.Cluster = "default"
+				}
+			}
+		}
+	}
+
+	return job, nil, nil
+}

nomad/job_endpoint_hook_consul_ce_test.go

@@ -0,0 +1,53 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+//go:build !ent
+// +build !ent
+
+package nomad
+
+import (
+	"testing"
+
+	"github.com/hashicorp/nomad/ci"
+	"github.com/hashicorp/nomad/nomad/mock"
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/hashicorp/nomad/testutil"
+	"github.com/shoenig/test"
+	"github.com/shoenig/test/must"
+)
+
+func TestJobEndpointHook_ConsulCE(t *testing.T) {
+	ci.Parallel(t)
+
+	srv, cleanup := TestServer(t, func(c *Config) {
+		c.NumSchedulers = 0
+	})
+	t.Cleanup(cleanup)
+	testutil.WaitForLeader(t, srv.RPC)
+
+	job := mock.Job()
+
+	// create two group-level services and assign to clusters
+	taskSvc := job.TaskGroups[0].Tasks[0].Services[0]
+	taskSvc.Provider = structs.ServiceProviderConsul
+	taskSvc.Cluster = "nondefault"
+	job.TaskGroups[0].Tasks[0].Services = []*structs.Service{taskSvc}
+
+	job.TaskGroups[0].Services = append(job.TaskGroups[0].Services, taskSvc.Copy())
+	job.TaskGroups[0].Services = append(job.TaskGroups[0].Services, taskSvc.Copy())
+	job.TaskGroups[0].Services[0].Cluster = ""
+	job.TaskGroups[0].Services[1].Cluster = "infra"
+
+	hook := jobConsulHook{srv}
+
+	_, _, err := hook.Mutate(job)
+
+	must.NoError(t, err)
+	test.Eq(t, "default", job.TaskGroups[0].Services[0].Cluster)
+	test.Eq(t, "infra", job.TaskGroups[0].Services[1].Cluster)
+	test.Eq(t, "nondefault", job.TaskGroups[0].Tasks[0].Services[0].Cluster)
+
+	_, err = hook.Validate(job)
+	must.EqError(t, err, "non-default Consul cluster requires Nomad Enterprise")
+}

nomad/job_endpoint_hook_vault.go

@@ -34,6 +34,11 @@ func (h jobVaultHook) Validate(job *structs.Job) ([]error, error) {
 		return nil, fmt.Errorf("Vault not enabled but used in the job")
 	}
 
+	err := h.validateClustersForNamespace(job, vaultBlocks)
+	if err != nil {
+		return nil, err
+	}
+
 	// Return early if Vault configuration doesn't require authentication.
 	if vconf.AllowsUnauthenticated() {
 		return nil, nil

nomad/job_endpoint_hook_vault_ce.go

@@ -7,6 +7,7 @@
 package nomad
 
 import (
+	"errors"
 	"fmt"
 	"strings"
 
@@ -27,3 +28,28 @@ func (jobVaultHook) validateNamespaces(
 	}
 	return nil
 }
+
+func (h jobVaultHook) validateClustersForNamespace(_ *structs.Job, blocks map[string]map[string]*structs.Vault) error {
+	for _, tg := range blocks {
+		for _, vault := range tg {
+			if vault.Cluster != "default" {
+				return errors.New("non-default Vault cluster requires Nomad Enterprise")
+			}
+		}
+	}
+
+	return nil
+}
+
+func (j jobVaultHook) Mutate(job *structs.Job) (*structs.Job, []error, error) {
+	for _, tg := range job.TaskGroups {
+		for _, task := range tg.Tasks {
+			if task.Vault == nil || task.Vault.Cluster != "" {
+				continue
+			}
+			task.Vault.Cluster = "default"
+		}
+	}
+
+	return job, nil, nil
+}

nomad/job_endpoint_hook_vault_ce_test.go

@@ -0,0 +1,47 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+//go:build !ent
+// +build !ent
+
+package nomad
+
+import (
+	"testing"
+
+	"github.com/hashicorp/nomad/ci"
+	"github.com/hashicorp/nomad/nomad/mock"
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/hashicorp/nomad/testutil"
+	"github.com/shoenig/test/must"
+)
+
+func TestJobEndpointHook_VaultCE(t *testing.T) {
+	ci.Parallel(t)
+
+	srv, cleanup := TestServer(t, func(c *Config) {
+		c.NumSchedulers = 0
+	})
+	t.Cleanup(cleanup)
+	testutil.WaitForLeader(t, srv.RPC)
+
+	job := mock.Job()
+
+	// create two different Vault blocks and assign to clusters
+	job.TaskGroups[0].Tasks = append(job.TaskGroups[0].Tasks, job.TaskGroups[0].Tasks[0].Copy())
+	job.TaskGroups[0].Tasks[0].Vault = &structs.Vault{Cluster: "default"}
+	job.TaskGroups[0].Tasks[1].Name = "web2"
+	job.TaskGroups[0].Tasks[1].Vault = &structs.Vault{Cluster: "infra"}
+
+	hook := jobVaultHook{srv}
+	_, _, err := hook.Mutate(job)
+	must.NoError(t, err)
+	must.Eq(t, "default", job.TaskGroups[0].Tasks[0].Vault.Cluster)
+	must.Eq(t, "infra", job.TaskGroups[0].Tasks[1].Vault.Cluster)
+
+	// skipping over the rest of Validate b/c it requires an actual
+	// Vault cluster
+	err = hook.validateClustersForNamespace(job, job.Vault())
+	must.EqError(t, err, "non-default Vault cluster requires Nomad Enterprise")
+
+}

nomad/mock/job.go

@@ -91,10 +91,12 @@ func Job() *structs.Job {
 										Timeout:  5 * time.Second,
 									},
 								},
+								Cluster: "default",
 							},
 							{
 								Name:      "${TASK}-admin",
 								PortLabel: "admin",
+								Cluster:   "default",
 							},
 						},
 						LogConfig: structs.DefaultLogConfig(),
@@ -227,10 +229,12 @@ func MultiTaskGroupJob() *structs.Job {
 								Timeout:  5 * time.Second,
 							},
 						},
+						Cluster: "default",
 					},
 					{
 						Name:      "${TASK}-admin",
 						PortLabel: "admin",
+						Cluster:   "default",
 					},
 				},
 				LogConfig: structs.DefaultLogConfig(),
@@ -553,10 +557,12 @@ func MaxParallelJob() *structs.Job {
 										Timeout:  5 * time.Second,
 									},
 								},
+								Cluster: "default",
 							},
 							{
 								Name:      "${TASK}-admin",
 								PortLabel: "admin",
+								Cluster:   "default",
 							},
 						},
 						LogConfig: structs.DefaultLogConfig(),