Job Register endpoint validates token

nomad/job_endpoint.go

@@ -2,6 +2,7 @@ package nomad
 
 import (
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/armon/go-metrics"
@@ -67,6 +68,42 @@ func (j *Job) Register(args *structs.JobRegisterRequest, reply *structs.JobRegis
 		}
 	}
 
+	// Ensure that the job has permissions for the requested Vault tokens
+	desiredPolicies := structs.VaultPoliciesSet(args.Job.VaultPolicies())
+	if len(desiredPolicies) != 0 {
+		vconf := j.srv.config.VaultConfig
+		if !vconf.Enabled {
+			return fmt.Errorf("Vault not enabled and Vault policies requested")
+		}
+
+		// Have to check if the user has permissions
+		if !vconf.AllowUnauthenticated {
+			if args.Job.VaultToken == "" {
+				return fmt.Errorf("Vault policies requested but missing Vault Token")
+			}
+
+			vault := j.srv.vault
+			s, err := vault.LookupToken(args.Job.VaultToken)
+			if err != nil {
+				return err
+			}
+
+			allowedPolicies, err := PoliciesFrom(s)
+			if err != nil {
+				return err
+			}
+
+			subset, offending := structs.SliceStringIsSubset(allowedPolicies, desiredPolicies)
+			if !subset {
+				return fmt.Errorf("Passed Vault Token doesn't allow access to the following policies: %s",
+					strings.Join(offending, ", "))
+			}
+		}
+	}
+
+	// Clear the Vault token
+	args.Job.VaultToken = ""
+
 	// Commit this update via Raft
 	_, index, err := j.srv.raftApply(structs.JobRegisterRequestType, args)
 	if err != nil {

nomad/job_endpoint_test.go

@@ -1,6 +1,7 @@
 package nomad
 
 import (
+	"fmt"
 	"reflect"
 	"strings"
 	"testing"
@@ -360,6 +361,189 @@ func TestJobEndpoint_Register_EnforceIndex(t *testing.T) {
 	}
 }
 
+func TestJobEndpoint_Register_Vault_Disabled(t *testing.T) {
+	s1 := testServer(t, func(c *Config) {
+		c.NumSchedulers = 0 // Prevent automatic dequeue
+		c.VaultConfig.Enabled = false
+	})
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	testutil.WaitForLeader(t, s1.RPC)
+
+	// Create the register request with a job asking for a vault policy
+	job := mock.Job()
+	job.TaskGroups[0].Tasks[0].Vault = &structs.Vault{Policies: []string{"foo"}}
+	req := &structs.JobRegisterRequest{
+		Job:          job,
+		WriteRequest: structs.WriteRequest{Region: "global"},
+	}
+
+	// Fetch the response
+	var resp structs.JobRegisterResponse
+	err := msgpackrpc.CallWithCodec(codec, "Job.Register", req, &resp)
+	if err == nil || !strings.Contains(err.Error(), "Vault not enabled") {
+		t.Fatalf("expected Vault not enabled error: %v", err)
+	}
+}
+
+func TestJobEndpoint_Register_Vault_AllowUnauthenticated(t *testing.T) {
+	s1 := testServer(t, func(c *Config) {
+		c.NumSchedulers = 0 // Prevent automatic dequeue
+	})
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	testutil.WaitForLeader(t, s1.RPC)
+
+	// Enable vault and allow authenticated
+	s1.config.VaultConfig.Enabled = true
+	s1.config.VaultConfig.AllowUnauthenticated = true
+
+	// Replace the Vault Client on the server
+	s1.vault = &TestVaultClient{}
+
+	// Create the register request with a job asking for a vault policy
+	job := mock.Job()
+	job.TaskGroups[0].Tasks[0].Vault = &structs.Vault{Policies: []string{"foo"}}
+	req := &structs.JobRegisterRequest{
+		Job:          job,
+		WriteRequest: structs.WriteRequest{Region: "global"},
+	}
+
+	// Fetch the response
+	var resp structs.JobRegisterResponse
+	err := msgpackrpc.CallWithCodec(codec, "Job.Register", req, &resp)
+	if err != nil {
+		t.Fatalf("bad: %v", err)
+	}
+
+	// Check for the job in the FSM
+	state := s1.fsm.State()
+	out, err := state.JobByID(job.ID)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if out == nil {
+		t.Fatalf("expected job")
+	}
+	if out.CreateIndex != resp.JobModifyIndex {
+		t.Fatalf("index mis-match")
+	}
+}
+
+func TestJobEndpoint_Register_Vault_NoToken(t *testing.T) {
+	s1 := testServer(t, func(c *Config) {
+		c.NumSchedulers = 0 // Prevent automatic dequeue
+	})
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	testutil.WaitForLeader(t, s1.RPC)
+
+	// Enable vault
+	s1.config.VaultConfig.Enabled = true
+	s1.config.VaultConfig.AllowUnauthenticated = false
+
+	// Replace the Vault Client on the server
+	s1.vault = &TestVaultClient{}
+
+	// Create the register request with a job asking for a vault policy but
+	// don't send a Vault token
+	job := mock.Job()
+	job.TaskGroups[0].Tasks[0].Vault = &structs.Vault{Policies: []string{"foo"}}
+	req := &structs.JobRegisterRequest{
+		Job:          job,
+		WriteRequest: structs.WriteRequest{Region: "global"},
+	}
+
+	// Fetch the response
+	var resp structs.JobRegisterResponse
+	err := msgpackrpc.CallWithCodec(codec, "Job.Register", req, &resp)
+	if err == nil || !strings.Contains(err.Error(), "missing Vault Token") {
+		t.Fatalf("expected Vault not enabled error: %v", err)
+	}
+}
+
+func TestJobEndpoint_Register_Vault_Policies(t *testing.T) {
+	s1 := testServer(t, func(c *Config) {
+		c.NumSchedulers = 0 // Prevent automatic dequeue
+	})
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	testutil.WaitForLeader(t, s1.RPC)
+
+	// Enable vault
+	s1.config.VaultConfig.Enabled = true
+	s1.config.VaultConfig.AllowUnauthenticated = false
+
+	// Replace the Vault Client on the server
+	tvc := &TestVaultClient{}
+	s1.vault = tvc
+
+	// Add three tokens: one that allows the requesting policy, one that does
+	// not and one that returns an error
+	policy := "foo"
+
+	badToken := structs.GenerateUUID()
+	badPolicies := []string{"a", "b", "c"}
+	tvc.SetLookupTokenAllowedPolicies(badToken, badPolicies)
+
+	goodToken := structs.GenerateUUID()
+	goodPolicies := []string{"foo", "bar", "baz"}
+	tvc.SetLookupTokenAllowedPolicies(goodToken, goodPolicies)
+
+	errToken := structs.GenerateUUID()
+	expectedErr := fmt.Errorf("return errors from vault")
+	tvc.SetLookupTokenError(errToken, expectedErr)
+
+	// Create the register request with a job asking for a vault policy but
+	// send the bad Vault token
+	job := mock.Job()
+	job.VaultToken = badToken
+	job.TaskGroups[0].Tasks[0].Vault = &structs.Vault{Policies: []string{policy}}
+	req := &structs.JobRegisterRequest{
+		Job:          job,
+		WriteRequest: structs.WriteRequest{Region: "global"},
+	}
+
+	// Fetch the response
+	var resp structs.JobRegisterResponse
+	err := msgpackrpc.CallWithCodec(codec, "Job.Register", req, &resp)
+	if err == nil || !strings.Contains(err.Error(),
+		"doesn't allow access to the following policies: "+policy) {
+		t.Fatalf("expected permission denied error: %v", err)
+	}
+
+	// Use the err token
+	job.VaultToken = errToken
+	err = msgpackrpc.CallWithCodec(codec, "Job.Register", req, &resp)
+	if err == nil || !strings.Contains(err.Error(), expectedErr.Error()) {
+		t.Fatalf("expected permission denied error: %v", err)
+	}
+
+	// Use the good token
+	job.VaultToken = goodToken
+
+	// Fetch the response
+	if err := msgpackrpc.CallWithCodec(codec, "Job.Register", req, &resp); err != nil {
+		t.Fatalf("bad: %v", err)
+	}
+
+	// Check for the job in the FSM
+	state := s1.fsm.State()
+	out, err := state.JobByID(job.ID)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	if out == nil {
+		t.Fatalf("expected job")
+	}
+	if out.CreateIndex != resp.JobModifyIndex {
+		t.Fatalf("index mis-match")
+	}
+	if out.VaultToken != "" {
+		t.Fatalf("vault token not cleared")
+	}
+}
+
 func TestJobEndpoint_Evaluate(t *testing.T) {
 	s1 := testServer(t, func(c *Config) {
 		c.NumSchedulers = 0 // Prevent automatic dequeue

nomad/structs/funcs.go

@@ -229,3 +229,44 @@ func CopySliceConstraints(s []*Constraint) []*Constraint {
 	}
 	return c
 }
+
+// SliceStringIsSubset returns whether the smaller set of strings is a subset of
+// the larger. If the smaller slice is not a subset, the offending elements are
+// returned.
+func SliceStringIsSubset(larger, smaller []string) (bool, []string) {
+	largerSet := make(map[string]struct{}, len(larger))
+	for _, l := range larger {
+		largerSet[l] = struct{}{}
+	}
+
+	subset := true
+	var offending []string
+	for _, s := range smaller {
+		if _, ok := largerSet[s]; !ok {
+			subset = false
+			offending = append(offending, s)
+		}
+	}
+
+	return subset, offending
+}
+
+// VaultPoliciesSet takes the structure returned by VaultPolicies and returns
+// the set of required policies
+func VaultPoliciesSet(policies map[string]map[string][]string) []string {
+	set := make(map[string]struct{})
+
+	for _, tgp := range policies {
+		for _, tp := range tgp {
+			for _, p := range tp {
+				set[p] = struct{}{}
+			}
+		}
+	}
+
+	flattened := make([]string, 0, len(set))
+	for p := range set {
+		flattened = append(flattened, p)
+	}
+	return flattened
+}

nomad/structs/funcs_test.go

@@ -235,3 +235,18 @@ func TestGenerateUUID(t *testing.T) {
 		}
 	}
 }
+
+func TestSliceStringIsSubset(t *testing.T) {
+	l := []string{"a", "b", "c"}
+	s := []string{"d"}
+
+	sub, offending := SliceStringIsSubset(l, l[:1])
+	if !sub || len(offending) != 0 {
+		t.Fatalf("bad %v %v", sub, offending)
+	}
+
+	sub, offending = SliceStringIsSubset(l, s)
+	if sub || len(offending) == 0 || offending[0] != "d" {
+		t.Fatalf("bad %v %v", sub, offending)
+	}
+}

nomad/structs/structs.go

@@ -1223,6 +1223,26 @@ func (j *Job) IsPeriodic() bool {
 	return j.Periodic != nil
 }
 
+// VaultPolicies returns the set of Vault policies per task group, per task
+func (j *Job) VaultPolicies() map[string]map[string][]string {
+	policies := make(map[string]map[string][]string, len(j.TaskGroups))
+
+	for _, tg := range j.TaskGroups {
+		tgPolicies := make(map[string][]string, len(tg.Tasks))
+		policies[tg.Name] = tgPolicies
+
+		for _, task := range tg.Tasks {
+			if task.Vault == nil {
+				continue
+			}
+
+			tgPolicies[task.Name] = task.Vault.Policies
+		}
+	}
+
+	return policies
+}
+
 // JobListStub is used to return a subset of job information
 // for the job list
 type JobListStub struct {

nomad/structs/structs_test.go

@@ -222,6 +222,86 @@ func TestJob_SystemJob_Validate(t *testing.T) {
 	}
 }
 
+func TestJob_VaultPolicies(t *testing.T) {
+	j0 := &Job{}
+	e0 := make(map[string]map[string][]string, 0)
+
+	j1 := &Job{
+		TaskGroups: []*TaskGroup{
+			&TaskGroup{
+				Name: "foo",
+				Tasks: []*Task{
+					&Task{
+						Name: "t1",
+					},
+					&Task{
+						Name: "t2",
+						Vault: &Vault{
+							Policies: []string{
+								"p1",
+								"p2",
+							},
+						},
+					},
+				},
+			},
+			&TaskGroup{
+				Name: "bar",
+				Tasks: []*Task{
+					&Task{
+						Name: "t3",
+						Vault: &Vault{
+							Policies: []string{
+								"p3",
+								"p4",
+							},
+						},
+					},
+					&Task{
+						Name: "t4",
+						Vault: &Vault{
+							Policies: []string{
+								"p5",
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	e1 := map[string]map[string][]string{
+		"foo": map[string][]string{
+			"t2": []string{"p1", "p2"},
+		},
+		"bar": map[string][]string{
+			"t3": []string{"p3", "p4"},
+			"t4": []string{"p5"},
+		},
+	}
+
+	cases := []struct {
+		Job      *Job
+		Expected map[string]map[string][]string
+	}{
+		{
+			Job:      j0,
+			Expected: e0,
+		},
+		{
+			Job:      j1,
+			Expected: e1,
+		},
+	}
+
+	for i, c := range cases {
+		got := c.Job.VaultPolicies()
+		if !reflect.DeepEqual(got, c.Expected) {
+			t.Fatalf("case %d: got %#v; want %#v", i+1, got, c.Expected)
+		}
+	}
+}
+
 func TestTaskGroup_Validate(t *testing.T) {
 	tg := &TaskGroup{
 		Count: -1,

nomad/vault_testing.go

@@ -0,0 +1,71 @@
+package nomad
+
+import (
+	"github.com/hashicorp/nomad/nomad/structs"
+	vapi "github.com/hashicorp/vault/api"
+)
+
+// TestVaultClient is a Vault client appropriate for use during testing. Its
+// behavior is programmable such that endpoints can be tested under various
+// circumstances.
+type TestVaultClient struct {
+	// LookupTokenErrors maps a token to an error that will be returned by the
+	// LookupToken call
+	LookupTokenErrors map[string]error
+
+	// LookupTokenSecret maps a token to the Vault secret that will be returned
+	// by the LookupToken call
+	LookupTokenSecret map[string]*vapi.Secret
+}
+
+func (v *TestVaultClient) LookupToken(token string) (*vapi.Secret, error) {
+	var secret *vapi.Secret
+	var err error
+
+	if v.LookupTokenSecret != nil {
+		secret = v.LookupTokenSecret[token]
+	}
+	if v.LookupTokenErrors != nil {
+		err = v.LookupTokenErrors[token]
+	}
+
+	return secret, err
+}
+
+// SetLookupTokenSecret sets the error that will be returned by the token
+// lookup
+func (v *TestVaultClient) SetLookupTokenError(token string, err error) {
+	if v.LookupTokenErrors == nil {
+		v.LookupTokenErrors = make(map[string]error)
+	}
+
+	v.LookupTokenErrors[token] = err
+}
+
+// SetLookupTokenSecret sets the secret that will be returned by the token
+// lookup
+func (v *TestVaultClient) SetLookupTokenSecret(token string, secret *vapi.Secret) {
+	if v.LookupTokenSecret == nil {
+		v.LookupTokenSecret = make(map[string]*vapi.Secret)
+	}
+
+	v.LookupTokenSecret[token] = secret
+}
+
+// SetLookupTokenAllowedPolicies is a helper that adds a secret that allows the
+// given policies
+func (v *TestVaultClient) SetLookupTokenAllowedPolicies(token string, policies []string) {
+	s := &vapi.Secret{
+		Data: map[string]interface{}{
+			"policies": policies,
+		},
+	}
+
+	v.SetLookupTokenSecret(token, s)
+}
+
+func (v *TestVaultClient) CreateToken(a *structs.Allocation, task string) (*vapi.Secret, error) {
+	return nil, nil
+}
+
+func (v *TestVaultClient) Stop() {}