nomad: implement ACL token endpoints

2026-01-08 03:15:42 +03:00 · 2017-08-12 15:44:05 -07:00
parent 0c8bb531cb
commit 14f7e8166f
5 changed files with 544 additions and 2 deletions
--- a/nomad/acl_endpoint.go
+++ b/nomad/acl_endpoint.go
@@ -148,3 +148,137 @@ func (a *ACL) GetPolicy(args *structs.ACLPolicySpecificRequest, reply *structs.S
 		}}
 	return a.srv.blockingRPC(&opts)
 }
+
+// UpsertTokens is used to create or update a set of tokens
+func (a *ACL) UpsertTokens(args *structs.ACLTokenUpsertRequest, reply *structs.GenericResponse) error {
+	if done, err := a.srv.forward("ACL.UpsertTokens", args, args, reply); done {
+		return err
+	}
+	defer metrics.MeasureSince([]string{"nomad", "acl", "upsert_tokens"}, time.Now())
+
+	// Validate non-zero set of tokens
+	if len(args.Tokens) == 0 {
+		return fmt.Errorf("must specify as least one token")
+	}
+
+	// Validate each token
+	for idx, token := range args.Tokens {
+		if err := token.Validate(); err != nil {
+			return fmt.Errorf("token %d invalid: %v", idx, err)
+		}
+	}
+
+	// Update via Raft
+	_, index, err := a.srv.raftApply(structs.ACLTokenUpsertRequestType, args)
+	if err != nil {
+		return err
+	}
+
+	// Update the index
+	reply.Index = index
+	return nil
+}
+
+// DeleteTokens is used to delete tokens
+func (a *ACL) DeleteTokens(args *structs.ACLTokenDeleteRequest, reply *structs.GenericResponse) error {
+	if done, err := a.srv.forward("ACL.DeleteTokens", args, args, reply); done {
+		return err
+	}
+	defer metrics.MeasureSince([]string{"nomad", "acl", "delete_tokens"}, time.Now())
+
+	// Validate non-zero set of tokens
+	if len(args.AccessorIDs) == 0 {
+		return fmt.Errorf("must specify as least one token")
+	}
+
+	// Update via Raft
+	_, index, err := a.srv.raftApply(structs.ACLTokenDeleteRequestType, args)
+	if err != nil {
+		return err
+	}
+
+	// Update the index
+	reply.Index = index
+	return nil
+}
+
+// ListTokens is used to list the tokens
+func (a *ACL) ListTokens(args *structs.ACLTokenListRequest, reply *structs.ACLTokenListResponse) error {
+	if done, err := a.srv.forward("ACL.ListTokens", args, args, reply); done {
+		return err
+	}
+	defer metrics.MeasureSince([]string{"nomad", "acl", "list_tokens"}, time.Now())
+
+	// Setup the blocking query
+	opts := blockingOptions{
+		queryOpts: &args.QueryOptions,
+		queryMeta: &reply.QueryMeta,
+		run: func(ws memdb.WatchSet, state *state.StateStore) error {
+			// Iterate over all the tokens
+			var err error
+			var iter memdb.ResultIterator
+			if prefix := args.QueryOptions.Prefix; prefix != "" {
+				iter, err = state.ACLTokenByPublicIDPrefix(ws, prefix)
+			} else {
+				iter, err = state.ACLTokens(ws)
+			}
+			if err != nil {
+				return err
+			}
+
+			// Convert all the tokens to a list stub
+			reply.Tokens = nil
+			for {
+				raw := iter.Next()
+				if raw == nil {
+					break
+				}
+				token := raw.(*structs.ACLToken)
+				reply.Tokens = append(reply.Tokens, token)
+			}
+
+			// Use the last index that affected the token table
+			index, err := state.Index("acl_token")
+			if err != nil {
+				return err
+			}
+			reply.Index = index
+			return nil
+		}}
+	return a.srv.blockingRPC(&opts)
+}
+
+// GetToken is used to get a specific token
+func (a *ACL) GetToken(args *structs.ACLTokenSpecificRequest, reply *structs.SingleACLTokenResponse) error {
+	if done, err := a.srv.forward("ACL.GetToken", args, args, reply); done {
+		return err
+	}
+	defer metrics.MeasureSince([]string{"nomad", "acl", "get_token"}, time.Now())
+
+	// Setup the blocking query
+	opts := blockingOptions{
+		queryOpts: &args.QueryOptions,
+		queryMeta: &reply.QueryMeta,
+		run: func(ws memdb.WatchSet, state *state.StateStore) error {
+			// Look for the token
+			out, err := state.ACLTokenByPublicID(ws, args.AccessorID)
+			if err != nil {
+				return err
+			}
+
+			// Setup the output
+			reply.Token = out
+			if out != nil {
+				reply.Index = out.ModifyIndex
+			} else {
+				// Use the last index that affected the token table
+				index, err := state.Index("acl_token")
+				if err != nil {
+					return err
+				}
+				reply.Index = index
+			}
+			return nil
+		}}
+	return a.srv.blockingRPC(&opts)
+}
--- a/nomad/acl_endpoint_test.go
+++ b/nomad/acl_endpoint_test.go
@@ -122,7 +122,7 @@ func TestACLEndpoint_GetPolicy_Blocking(t *testing.T) {
 	}
 }

-func TestACLEndpoint_List(t *testing.T) {
+func TestACLEndpoint_ListPolicies(t *testing.T) {
 	t.Parallel()
 	s1 := testServer(t, nil)
 	defer s1.Shutdown()
@@ -163,7 +163,7 @@ func TestACLEndpoint_List(t *testing.T) {
 	assert.Equal(t, 1, len(resp2.Policies))
 }

-func TestACLEndpoint_List_Blocking(t *testing.T) {
+func TestACLEndpoint_ListPolicies_Blocking(t *testing.T) {
 	t.Parallel()
 	s1 := testServer(t, nil)
 	defer s1.Shutdown()
@@ -295,3 +295,287 @@ func TestACLEndpoint_UpsertPolicies_Invalid(t *testing.T) {
 		t.Fatalf("bad: %s", err)
 	}
 }
+
+func TestACLEndpoint_GetToken(t *testing.T) {
+	t.Parallel()
+	s1 := testServer(t, nil)
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	testutil.WaitForLeader(t, s1.RPC)
+
+	// Create the register request
+	token := mock.ACLToken()
+	s1.fsm.State().UpsertACLTokens(1000, []*structs.ACLToken{token})
+
+	// Lookup the token
+	get := &structs.ACLTokenSpecificRequest{
+		AccessorID:   token.AccessorID,
+		QueryOptions: structs.QueryOptions{Region: "global"},
+	}
+	var resp structs.SingleACLTokenResponse
+	if err := msgpackrpc.CallWithCodec(codec, "ACL.GetToken", get, &resp); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	assert.Equal(t, uint64(1000), resp.Index)
+	assert.Equal(t, token, resp.Token)
+
+	// Lookup non-existing token
+	get.AccessorID = structs.GenerateUUID()
+	if err := msgpackrpc.CallWithCodec(codec, "ACL.GetToken", get, &resp); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	assert.Equal(t, uint64(1000), resp.Index)
+	assert.Nil(t, resp.Token)
+}
+
+func TestACLEndpoint_GetToken_Blocking(t *testing.T) {
+	t.Parallel()
+	s1 := testServer(t, nil)
+	defer s1.Shutdown()
+	state := s1.fsm.State()
+	codec := rpcClient(t, s1)
+	testutil.WaitForLeader(t, s1.RPC)
+
+	// Create the tokens
+	p1 := mock.ACLToken()
+	p2 := mock.ACLToken()
+
+	// First create an unrelated token
+	time.AfterFunc(100*time.Millisecond, func() {
+		err := state.UpsertACLTokens(100, []*structs.ACLToken{p1})
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	})
+
+	// Upsert the token we are watching later
+	time.AfterFunc(200*time.Millisecond, func() {
+		err := state.UpsertACLTokens(200, []*structs.ACLToken{p2})
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	})
+
+	// Lookup the token
+	req := &structs.ACLTokenSpecificRequest{
+		AccessorID: p2.AccessorID,
+		QueryOptions: structs.QueryOptions{
+			Region:        "global",
+			MinQueryIndex: 150,
+		},
+	}
+	var resp structs.SingleACLTokenResponse
+	start := time.Now()
+	if err := msgpackrpc.CallWithCodec(codec, "ACL.GetToken", req, &resp); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	if elapsed := time.Since(start); elapsed < 200*time.Millisecond {
+		t.Fatalf("should block (returned in %s) %#v", elapsed, resp)
+	}
+	if resp.Index != 200 {
+		t.Fatalf("Bad index: %d %d", resp.Index, 200)
+	}
+	if resp.Token == nil || resp.Token.AccessorID != p2.AccessorID {
+		t.Fatalf("bad: %#v", resp.Token)
+	}
+
+	// Eval delete triggers watches
+	time.AfterFunc(100*time.Millisecond, func() {
+		err := state.DeleteACLTokens(300, []string{p2.AccessorID})
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	})
+
+	req.QueryOptions.MinQueryIndex = 250
+	var resp2 structs.SingleACLTokenResponse
+	start = time.Now()
+	if err := msgpackrpc.CallWithCodec(codec, "ACL.GetToken", req, &resp2); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	if elapsed := time.Since(start); elapsed < 100*time.Millisecond {
+		t.Fatalf("should block (returned in %s) %#v", elapsed, resp2)
+	}
+	if resp2.Index != 300 {
+		t.Fatalf("Bad index: %d %d", resp2.Index, 300)
+	}
+	if resp2.Token != nil {
+		t.Fatalf("bad: %#v", resp2.Token)
+	}
+}
+
+func TestACLEndpoint_ListTokens(t *testing.T) {
+	t.Parallel()
+	s1 := testServer(t, nil)
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	testutil.WaitForLeader(t, s1.RPC)
+
+	// Create the register request
+	p1 := mock.ACLToken()
+	p2 := mock.ACLToken()
+
+	p1.AccessorID = "aaaaaaaa-3350-4b4b-d185-0e1992ed43e9"
+	p2.AccessorID = "aaaabbbb-3350-4b4b-d185-0e1992ed43e9"
+	s1.fsm.State().UpsertACLTokens(1000, []*structs.ACLToken{p1, p2})
+
+	// Lookup the tokens
+	get := &structs.ACLTokenListRequest{
+		QueryOptions: structs.QueryOptions{Region: "global"},
+	}
+	var resp structs.ACLTokenListResponse
+	if err := msgpackrpc.CallWithCodec(codec, "ACL.ListTokens", get, &resp); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	assert.Equal(t, uint64(1000), resp.Index)
+	assert.Equal(t, 2, len(resp.Tokens))
+
+	// Lookup the tokens by prefix
+	get = &structs.ACLTokenListRequest{
+		QueryOptions: structs.QueryOptions{
+			Region: "global",
+			Prefix: "aaaabb",
+		},
+	}
+	var resp2 structs.ACLTokenListResponse
+	if err := msgpackrpc.CallWithCodec(codec, "ACL.ListTokens", get, &resp2); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	assert.Equal(t, uint64(1000), resp2.Index)
+	assert.Equal(t, 1, len(resp2.Tokens))
+}
+
+func TestACLEndpoint_ListTokens_Blocking(t *testing.T) {
+	t.Parallel()
+	s1 := testServer(t, nil)
+	defer s1.Shutdown()
+	state := s1.fsm.State()
+	codec := rpcClient(t, s1)
+	testutil.WaitForLeader(t, s1.RPC)
+
+	// Create the token
+	token := mock.ACLToken()
+
+	// Upsert eval triggers watches
+	time.AfterFunc(100*time.Millisecond, func() {
+		if err := state.UpsertACLTokens(2, []*structs.ACLToken{token}); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	})
+
+	req := &structs.ACLTokenListRequest{
+		QueryOptions: structs.QueryOptions{
+			Region:        "global",
+			MinQueryIndex: 1,
+		},
+	}
+	start := time.Now()
+	var resp structs.ACLTokenListResponse
+	if err := msgpackrpc.CallWithCodec(codec, "ACL.ListTokens", req, &resp); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	if elapsed := time.Since(start); elapsed < 100*time.Millisecond {
+		t.Fatalf("should block (returned in %s) %#v", elapsed, resp)
+	}
+	assert.Equal(t, uint64(2), resp.Index)
+	if len(resp.Tokens) != 1 || resp.Tokens[0].AccessorID != token.AccessorID {
+		t.Fatalf("bad: %#v", resp.Tokens)
+	}
+
+	// Eval deletion triggers watches
+	time.AfterFunc(100*time.Millisecond, func() {
+		if err := state.DeleteACLTokens(3, []string{token.AccessorID}); err != nil {
+			t.Fatalf("err: %v", err)
+		}
+	})
+
+	req.MinQueryIndex = 2
+	start = time.Now()
+	var resp2 structs.ACLTokenListResponse
+	if err := msgpackrpc.CallWithCodec(codec, "ACL.ListTokens", req, &resp2); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	if elapsed := time.Since(start); elapsed < 100*time.Millisecond {
+		t.Fatalf("should block (returned in %s) %#v", elapsed, resp2)
+	}
+	assert.Equal(t, uint64(3), resp2.Index)
+	assert.Equal(t, 0, len(resp2.Tokens))
+}
+
+func TestACLEndpoint_DeleteTokens(t *testing.T) {
+	t.Parallel()
+	s1 := testServer(t, nil)
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	testutil.WaitForLeader(t, s1.RPC)
+
+	// Create the register request
+	p1 := mock.ACLToken()
+	s1.fsm.State().UpsertACLTokens(1000, []*structs.ACLToken{p1})
+
+	// Lookup the tokens
+	req := &structs.ACLTokenDeleteRequest{
+		AccessorIDs:  []string{p1.AccessorID},
+		WriteRequest: structs.WriteRequest{Region: "global"},
+	}
+	var resp structs.GenericResponse
+	if err := msgpackrpc.CallWithCodec(codec, "ACL.DeleteTokens", req, &resp); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	assert.NotEqual(t, uint64(0), resp.Index)
+}
+
+func TestACLEndpoint_UpsertTokens(t *testing.T) {
+	t.Parallel()
+	s1 := testServer(t, nil)
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	testutil.WaitForLeader(t, s1.RPC)
+
+	// Create the register request
+	p1 := mock.ACLToken()
+
+	// Lookup the tokens
+	req := &structs.ACLTokenUpsertRequest{
+		Tokens:       []*structs.ACLToken{p1},
+		WriteRequest: structs.WriteRequest{Region: "global"},
+	}
+	var resp structs.GenericResponse
+	if err := msgpackrpc.CallWithCodec(codec, "ACL.UpsertTokens", req, &resp); err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	assert.NotEqual(t, uint64(0), resp.Index)
+
+	// Check we created the token
+	out, err := s1.fsm.State().ACLTokenByPublicID(nil, p1.AccessorID)
+	assert.Nil(t, err)
+	assert.NotNil(t, out)
+}
+
+func TestACLEndpoint_UpsertTokens_Invalid(t *testing.T) {
+	t.Parallel()
+	s1 := testServer(t, nil)
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	testutil.WaitForLeader(t, s1.RPC)
+
+	// Create the register request
+	p1 := mock.ACLToken()
+	p1.Type = "blah blah"
+
+	// Lookup the tokens
+	req := &structs.ACLTokenUpsertRequest{
+		Tokens:       []*structs.ACLToken{p1},
+		WriteRequest: structs.WriteRequest{Region: "global"},
+	}
+	var resp structs.GenericResponse
+	err := msgpackrpc.CallWithCodec(codec, "ACL.UpsertTokens", req, &resp)
+	assert.NotNil(t, err)
+	if !strings.Contains(err.Error(), "client or management") {
+		t.Fatalf("bad: %s", err)
+	}
+}
--- a/nomad/fsm.go
+++ b/nomad/fsm.go
@@ -173,6 +173,10 @@ func (n *nomadFSM) Apply(log *raft.Log) interface{} {
 		return n.applyACLPolicyUpsert(buf[1:], log.Index)
 	case structs.ACLPolicyDeleteRequestType:
 		return n.applyACLPolicyDelete(buf[1:], log.Index)
+	case structs.ACLTokenUpsertRequestType:
+		return n.applyACLTokenUpsert(buf[1:], log.Index)
+	case structs.ACLTokenDeleteRequestType:
+		return n.applyACLTokenDelete(buf[1:], log.Index)
 	default:
 		if ignoreUnknown {
 			n.logger.Printf("[WARN] nomad.fsm: ignoring unknown message type (%d), upgrade to newer version", msgType)
@@ -705,6 +709,36 @@ func (n *nomadFSM) applyACLPolicyDelete(buf []byte, index uint64) interface{} {
 	return nil
 }

+// applyACLTokenUpsert is used to upsert a set of policies
+func (n *nomadFSM) applyACLTokenUpsert(buf []byte, index uint64) interface{} {
+	defer metrics.MeasureSince([]string{"nomad", "fsm", "apply_acl_token_upsert"}, time.Now())
+	var req structs.ACLTokenUpsertRequest
+	if err := structs.Decode(buf, &req); err != nil {
+		panic(fmt.Errorf("failed to decode request: %v", err))
+	}
+
+	if err := n.state.UpsertACLTokens(index, req.Tokens); err != nil {
+		n.logger.Printf("[ERR] nomad.fsm: UpsertACLTokens failed: %v", err)
+		return err
+	}
+	return nil
+}
+
+// applyACLTokenDelete is used to delete a set of policies
+func (n *nomadFSM) applyACLTokenDelete(buf []byte, index uint64) interface{} {
+	defer metrics.MeasureSince([]string{"nomad", "fsm", "apply_acl_token_delete"}, time.Now())
+	var req structs.ACLTokenDeleteRequest
+	if err := structs.Decode(buf, &req); err != nil {
+		panic(fmt.Errorf("failed to decode request: %v", err))
+	}
+
+	if err := n.state.DeleteACLTokens(index, req.AccessorIDs); err != nil {
+		n.logger.Printf("[ERR] nomad.fsm: DeleteACLTokens failed: %v", err)
+		return err
+	}
+	return nil
+}
+
 func (n *nomadFSM) Snapshot() (raft.FSMSnapshot, error) {
 	// Create a new snapshot
 	snap, err := n.state.Snapshot()
--- a/nomad/fsm_test.go
+++ b/nomad/fsm_test.go
@@ -1571,6 +1571,59 @@ func TestFSM_DeleteACLPolicies(t *testing.T) {
 	assert.Nil(t, out)
 }

+func TestFSM_UpsertACLTokens(t *testing.T) {
+	t.Parallel()
+	fsm := testFSM(t)
+
+	token := mock.ACLToken()
+	req := structs.ACLTokenUpsertRequest{
+		Tokens: []*structs.ACLToken{token},
+	}
+	buf, err := structs.Encode(structs.ACLTokenUpsertRequestType, req)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	resp := fsm.Apply(makeLog(buf))
+	if resp != nil {
+		t.Fatalf("resp: %v", resp)
+	}
+
+	// Verify we are registered
+	ws := memdb.NewWatchSet()
+	out, err := fsm.State().ACLTokenByPublicID(ws, token.AccessorID)
+	assert.Nil(t, err)
+	assert.NotNil(t, out)
+}
+
+func TestFSM_DeleteACLTokens(t *testing.T) {
+	t.Parallel()
+	fsm := testFSM(t)
+
+	token := mock.ACLToken()
+	err := fsm.State().UpsertACLTokens(1000, []*structs.ACLToken{token})
+	assert.Nil(t, err)
+
+	req := structs.ACLTokenDeleteRequest{
+		AccessorIDs: []string{token.AccessorID},
+	}
+	buf, err := structs.Encode(structs.ACLTokenDeleteRequestType, req)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+
+	resp := fsm.Apply(makeLog(buf))
+	if resp != nil {
+		t.Fatalf("resp: %v", resp)
+	}
+
+	// Verify we are NOT registered
+	ws := memdb.NewWatchSet()
+	out, err := fsm.State().ACLTokenByPublicID(ws, token.AccessorID)
+	assert.Nil(t, err)
+	assert.Nil(t, out)
+}
+
 func testSnapshotRestore(t *testing.T, fsm *nomadFSM) *nomadFSM {
 	// Snapshot
 	snap, err := fsm.Snapshot()
--- a/nomad/structs/structs.go
+++ b/nomad/structs/structs.go
@@ -65,6 +65,8 @@ const (
 	JobStabilityRequestType
 	ACLPolicyUpsertRequestType
 	ACLPolicyDeleteRequestType
+	ACLTokenUpsertRequestType
+	ACLTokenDeleteRequestType
 )

 const (
@@ -5452,3 +5454,38 @@ func (a *ACLToken) Validate() error {
 	}
 	return mErr.ErrorOrNil()
 }
+
+// ACLTokenListRequest is used to request a list of tokens
+type ACLTokenListRequest struct {
+	QueryOptions
+}
+
+// ACLTokenSpecificRequest is used to query a specific token
+type ACLTokenSpecificRequest struct {
+	AccessorID string
+	QueryOptions
+}
+
+// ACLTokenListResponse is used for a list request
+type ACLTokenListResponse struct {
+	Tokens []*ACLToken
+	QueryMeta
+}
+
+// SingleACLTokenResponse is used to return a single token
+type SingleACLTokenResponse struct {
+	Token *ACLToken
+	QueryMeta
+}
+
+// ACLTokenDeleteRequest is used to delete a set of tokens
+type ACLTokenDeleteRequest struct {
+	AccessorIDs []string
+	WriteRequest
+}
+
+// ACLTokenUpsertRequest is used to upsert a set of tokens
+type ACLTokenUpsertRequest struct {
+	Tokens []*ACLToken
+	WriteRequest
+}