Add the Read/List service functions (#2123)

This commit is contained in:
Todd authored and GitHub committed 2022-06-03 11:19:04 -07:00
1 parent 5c76017dd6
commit abbfe45419
9 files changed
+747 -108

No files matched your search

+11
View File
@@ -35,6 +35,7 @@ import (
"github.com/hashicorp/boundary/internal/daemon/controller/handlers/sessions"
"github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets"
"github.com/hashicorp/boundary/internal/daemon/controller/handlers/users"
"github.com/hashicorp/boundary/internal/daemon/controller/handlers/workers"
"github.com/hashicorp/boundary/internal/daemon/controller/internal/metric"
"github.com/hashicorp/boundary/internal/gen/controller/api/services"
authpb "github.com/hashicorp/boundary/internal/gen/controller/auth"
@@ -224,6 +225,13 @@ func (c *Controller) registerGrpcServices(s *grpc.Server) error {
}
services.RegisterCredentialLibraryServiceServer(s, cl)
}
if _, ok := currentServices[services.WorkerService_ServiceDesc.ServiceName]; !ok {
ws, err := workers.NewService(c.baseContext, c.ServersRepoFn, c.IamRepoFn)
if err != nil {
return fmt.Errorf("failed to create worker handler service: %w", err)
}
services.RegisterWorkerServiceServer(s, ws)
}
if _, ok := s.GetServiceInfo()[opsservices.HealthService_ServiceDesc.ServiceName]; !ok {
hs := health.NewService()
opsservices.RegisterHealthServiceServer(s, hs)
@@ -281,6 +289,9 @@ func registerGrpcGatewayEndpoints(ctx context.Context, gwMux *runtime.ServeMux,
if err := services.RegisterCredentialLibraryServiceHandlerFromEndpoint(ctx, gwMux, gatewayTarget, dialOptions); err != nil {
return fmt.Errorf("failed to register credential library service handler: %w", err)
}
if err := services.RegisterWorkerServiceHandlerFromEndpoint(ctx, gwMux, gatewayTarget, dialOptions); err != nil {
return fmt.Errorf("failed to register worker service handler: %w", err)
}
return nil
}
@@ -0,0 +1,359 @@
package workers
import (
"context"
"fmt"
"github.com/hashicorp/boundary/globals"
"github.com/hashicorp/boundary/internal/daemon/controller/auth"
"github.com/hashicorp/boundary/internal/daemon/controller/common"
"github.com/hashicorp/boundary/internal/daemon/controller/common/scopeids"
"github.com/hashicorp/boundary/internal/daemon/controller/handlers"
"github.com/hashicorp/boundary/internal/errors"
pbs "github.com/hashicorp/boundary/internal/gen/controller/api/services"
"github.com/hashicorp/boundary/internal/perms"
"github.com/hashicorp/boundary/internal/requests"
"github.com/hashicorp/boundary/internal/servers"
"github.com/hashicorp/boundary/internal/types/action"
"github.com/hashicorp/boundary/internal/types/resource"
"github.com/hashicorp/boundary/internal/types/scope"
pb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/workers"
"google.golang.org/grpc/codes"
"google.golang.org/protobuf/types/known/structpb"
"google.golang.org/protobuf/types/known/wrapperspb"
)
var (
// IdActions contains the set of actions that can be performed on
// individual resources
IdActions = action.ActionSet{
action.NoOp,
action.Read,
action.Update,
action.Delete,
}
// CollectionActions contains the set of actions that can be performed on
// this collection
CollectionActions = action.ActionSet{
action.Create,
action.List,
}
)
// Service handles request as described by the pbs.WorkerServiceServer interface.
type Service struct {
pbs.UnimplementedWorkerServiceServer
repoFn common.ServersRepoFactory
iamRepoFn common.IamRepoFactory
}
// NewService returns a worker service which handles worker related requests to boundary.
func NewService(ctx context.Context, repo common.ServersRepoFactory, iamRepoFn common.IamRepoFactory) (Service, error) {
const op = "workers.NewService"
if repo == nil {
return Service{}, errors.New(ctx, errors.InvalidParameter, op, "missing servers repository")
}
if iamRepoFn == nil {
return Service{}, errors.New(ctx, errors.InvalidParameter, op, "missing iam repository")
}
return Service{repoFn: repo, iamRepoFn: iamRepoFn}, nil
}
var _ pbs.WorkerServiceServer = Service{}
// ListWorkers implements the interface pbs.WorkerServiceServer.
func (s Service) ListWorkers(ctx context.Context, req *pbs.ListWorkersRequest) (*pbs.ListWorkersResponse, error) {
if err := validateListRequest(req); err != nil {
return nil, err
}
authResults := s.authResult(ctx, req.GetScopeId(), action.List)
if authResults.Error != nil {
// If it's forbidden, and it's a recursive request, and they're
// successfully authenticated but just not authorized, keep going as we
// may have authorization on downstream scopes. Or, if they've not
// authenticated, still process in case u_anon has permissions.
if (authResults.Error == handlers.ForbiddenError() || authResults.Error == handlers.UnauthenticatedError()) &&
req.GetRecursive() &&
authResults.AuthenticationFinished {
} else {
return nil, authResults.Error
}
}
scopeIds, scopeInfoMap, err := scopeids.GetListingScopeIds(
ctx, s.iamRepoFn, authResults, req.GetScopeId(), resource.Worker, req.GetRecursive())
if err != nil {
return nil, err
}
// If no scopes match, return an empty response
if len(scopeIds) == 0 {
return &pbs.ListWorkersResponse{}, nil
}
ul, err := s.listFromRepo(ctx, scopeIds)
if err != nil {
return nil, err
}
if len(ul) == 0 {
return &pbs.ListWorkersResponse{}, nil
}
filter, err := handlers.NewFilter(req.GetFilter())
if err != nil {
return nil, err
}
finalItems := make([]*pb.Worker, 0, len(ul))
res := perms.Resource{
Type: resource.Worker,
}
for _, item := range ul {
res.Id = item.GetPublicId()
res.ScopeId = item.GetScopeId()
authorizedActions := authResults.FetchActionSetForId(ctx, item.GetPublicId(), IdActions, auth.WithResource(&res)).Strings()
if len(authorizedActions) == 0 {
continue
}
outputFields := authResults.FetchOutputFields(res, action.List).SelfOrDefaults(authResults.UserId)
outputOpts := make([]handlers.Option, 0, 3)
outputOpts = append(outputOpts, handlers.WithOutputFields(&outputFields))
if outputFields.Has(globals.ScopeField) {
outputOpts = append(outputOpts, handlers.WithScope(scopeInfoMap[item.GetScopeId()]))
}
if outputFields.Has(globals.AuthorizedActionsField) {
outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authorizedActions))
}
item, err := toProto(ctx, item, outputOpts...)
if err != nil {
return nil, err
}
if filter.Match(item) {
finalItems = append(finalItems, item)
}
}
return &pbs.ListWorkersResponse{Items: finalItems}, nil
}
// GetWorkers implements the interface pbs.WorkerServiceServer.
func (s Service) GetWorker(ctx context.Context, req *pbs.GetWorkerRequest) (*pbs.GetWorkerResponse, error) {
const op = "workers.(Service).GetWorker"
if err := validateGetRequest(req); err != nil {
return nil, err
}
authResults := s.authResult(ctx, req.GetId(), action.Read)
if authResults.Error != nil {
return nil, authResults.Error
}
w, err := s.getFromRepo(ctx, req.GetId())
if err != nil {
return nil, err
}
outputFields, ok := requests.OutputFields(ctx)
if !ok {
return nil, errors.New(ctx, errors.Internal, op, "no request context found")
}
outputOpts := make([]handlers.Option, 0, 3)
outputOpts = append(outputOpts, handlers.WithOutputFields(&outputFields))
if outputFields.Has(globals.ScopeField) {
outputOpts = append(outputOpts, handlers.WithScope(authResults.Scope))
}
if outputFields.Has(globals.AuthorizedActionsField) {
outputOpts = append(outputOpts, handlers.WithAuthorizedActions(authResults.FetchActionSetForId(ctx, w.GetPublicId(), IdActions).Strings()))
}
item, err := toProto(ctx, w, outputOpts...)
if err != nil {
return nil, err
}
return &pbs.GetWorkerResponse{Item: item}, nil
}
func (s Service) listFromRepo(ctx context.Context, scopeIds []string) ([]*servers.Worker, error) {
repo, err := s.repoFn()
if err != nil {
return nil, err
}
wl, err := repo.ListWorkers(ctx, scopeIds, servers.WithLiveness(-1))
if err != nil {
return nil, err
}
return wl, nil
}
func (s Service) getFromRepo(ctx context.Context, id string) (*servers.Worker, error) {
repo, err := s.repoFn()
if err != nil {
return nil, err
}
w, err := repo.LookupWorker(ctx, id)
if err != nil {
if errors.IsNotFoundError(err) {
return nil, handlers.NotFoundErrorf("Worker %q doesn't exist.", id)
}
return nil, err
}
if w == nil {
return nil, handlers.NotFoundErrorf("Worker %q doesn't exist.", id)
}
return w, nil
}
func (s Service) authResult(ctx context.Context, id string, a action.Type) auth.VerifyResults {
res := auth.VerifyResults{}
repo, err := s.repoFn()
if err != nil {
res.Error = err
return res
}
var parentId string
opts := []auth.Option{auth.WithType(resource.Worker), auth.WithAction(a)}
switch a {
case action.List, action.Create:
parentId = id
default:
w, err := repo.LookupWorker(ctx, id)
if err != nil {
res.Error = err
return res
}
if w == nil {
res.Error = handlers.NotFoundError()
return res
}
parentId = w.GetScopeId()
opts = append(opts, auth.WithId(id))
}
opts = append(opts, auth.WithScopeId(parentId))
return auth.Verify(ctx, opts...)
}
func toProto(ctx context.Context, in *servers.Worker, opt ...handlers.Option) (*pb.Worker, error) {
const op = "workers.toProto"
opts := handlers.GetOpts(opt...)
if opts.WithOutputFields == nil {
return nil, handlers.ApiErrorWithCodeAndMessage(codes.Internal, "output fields not found when building worker proto")
}
outputFields := *opts.WithOutputFields
out := pb.Worker{}
if outputFields.Has(globals.IdField) {
out.Id = in.GetPublicId()
}
if outputFields.Has(globals.ScopeIdField) {
out.ScopeId = in.GetScopeId()
}
if outputFields.Has(globals.DescriptionField) && in.GetDescription() != "" {
out.Description = wrapperspb.String(in.GetDescription())
}
if outputFields.Has(globals.NameField) && in.GetName() != "" {
out.Name = wrapperspb.String(in.GetName())
}
if outputFields.Has(globals.CreatedTimeField) {
out.CreatedTime = in.GetCreateTime().GetTimestamp()
}
if outputFields.Has(globals.UpdatedTimeField) {
out.UpdatedTime = in.GetUpdateTime().GetTimestamp()
}
if outputFields.Has(globals.VersionField) {
out.Version = in.GetVersion()
}
if outputFields.Has(globals.ScopeField) {
out.Scope = opts.WithScope
}
if outputFields.Has(globals.AuthorizedActionsField) {
out.AuthorizedActions = opts.WithAuthorizedActions
}
if outputFields.Has(globals.AddressField) && in.GetAddress() != "" {
out.Address = wrapperspb.String(in.GetAddress())
}
if outputFields.Has(globals.CanonicalAddressField) {
out.CanonicalAddress = in.CanonicalAddress()
}
if outputFields.Has(globals.LastStatusTimeField) {
out.LastStatusTime = in.GetLastStatusTime().GetTimestamp()
}
if outputFields.Has(globals.CanonicalTagsField) && len(in.CanonicalTags()) > 0 {
var err error
out.Tags, err = tagsToMapProto(in.CanonicalTags())
if err != nil {
return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error preparing canonical tags proto"))
}
}
if outputFields.Has(globals.TagsField) && len(in.GetApiTags()) > 0 {
var err error
out.Tags, err = tagsToMapProto(in.GetApiTags())
if err != nil {
return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error preparing api tags proto"))
}
}
if outputFields.Has(globals.ConfigurationField) {
if in.GetWorkerReportedAddress() != "" ||
in.GetWorkerReportedName() != "" ||
len(in.GetConfigTags()) > 0 {
out.WorkerConfig = &pb.WorkerConfig{}
}
if len(in.GetConfigTags()) > 0 {
var err error
out.GetWorkerConfig().Tags, err = tagsToMapProto(in.GetConfigTags())
if err != nil {
return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error preparing config tags proto"))
}
}
out.GetWorkerConfig().Address = in.GetWorkerReportedAddress()
out.GetWorkerConfig().Name = in.GetWorkerReportedName()
}
return &out, nil
}
func tagsToMapProto(in map[string][]string) (map[string]*structpb.ListValue, error) {
b := make(map[string][]interface{})
for k, v := range in {
result := make([]interface{}, 0, len(v))
for _, t := range v {
result = append(result, t)
}
b[k] = result
}
ret := make(map[string]*structpb.ListValue)
var err error
for k, v := range b {
ret[k], err = structpb.NewList(v)
if err != nil {
return nil, err
}
}
return ret, nil
}
// A validateX method should exist for each method above. These methods do not make calls to any backing service but enforce
// requirements on the structure of the request. They verify that:
// * The path passed in is correctly formatted
// * All required parameters are set
// * There are no conflicting parameters provided
func validateGetRequest(req *pbs.GetWorkerRequest) error {
return handlers.ValidateGetRequest(handlers.NoopValidatorFn, req, servers.WorkerPrefix)
}
func validateListRequest(req *pbs.ListWorkersRequest) error {
badFields := map[string]string{}
if req.GetScopeId() != scope.Global.String() {
badFields["scope_id"] = "Must be 'global' when listing."
}
if _, err := handlers.NewFilter(req.GetFilter()); err != nil {
badFields["filter"] = fmt.Sprintf("This field could not be parsed. %v", err)
}
if len(badFields) > 0 {
return handlers.InvalidArgumentErrorf("Error in provided request.", badFields)
}
return nil
}
@@ -0,0 +1,223 @@
package workers
import (
"context"
"fmt"
"sort"
"testing"
"github.com/google/go-cmp/cmp"
"github.com/hashicorp/boundary/internal/daemon/controller/auth"
"github.com/hashicorp/boundary/internal/daemon/controller/handlers"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/errors"
pbs "github.com/hashicorp/boundary/internal/gen/controller/api/services"
"github.com/hashicorp/boundary/internal/iam"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/servers"
"github.com/hashicorp/boundary/internal/types/scope"
"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/scopes"
pb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/workers"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/grpc/codes"
"google.golang.org/protobuf/testing/protocmp"
"google.golang.org/protobuf/types/known/wrapperspb"
)
var testAuthorizedActions = []string{"no-op", "read", "update", "delete"}
func TestGet(t *testing.T) {
conn, _ := db.TestSetup(t, "postgres")
wrap := db.TestWrapper(t)
iamRepo := iam.TestRepo(t, conn, wrap)
iamRepoFn := func() (*iam.Repository, error) {
return iamRepo, nil
}
rw := db.New(conn)
kms := kms.TestKms(t, conn, wrap)
repoFn := func() (*servers.Repository, error) {
return servers.NewRepository(rw, rw, kms)
}
worker := servers.TestWorker(t, conn, wrap,
servers.WithName("test worker names"),
servers.WithDescription("test worker description"),
servers.WithAddress("test worker address"))
wantWorker := &pb.Worker{
Id: worker.GetPublicId(),
ScopeId: worker.GetScopeId(),
Scope: &scopes.ScopeInfo{Id: worker.GetScopeId(), Type: scope.Global.String(), Name: scope.Global.String(), Description: "Global Scope"},
CreatedTime: worker.CreateTime.GetTimestamp(),
UpdatedTime: worker.UpdateTime.GetTimestamp(),
Version: worker.GetVersion(),
Name: wrapperspb.String(worker.GetName()),
Description: wrapperspb.String(worker.GetDescription()),
Address: wrapperspb.String(worker.GetAddress()),
AuthorizedActions: testAuthorizedActions,
CanonicalAddress: worker.CanonicalAddress(),
LastStatusTime: worker.GetLastStatusTime().GetTimestamp(),
WorkerConfig: &pb.WorkerConfig{
Address: worker.GetWorkerReportedAddress(),
Name: worker.GetWorkerReportedName(),
},
}
cases := []struct {
name string
scopeId string
req *pbs.GetWorkerRequest
res *pbs.GetWorkerResponse
err error
}{
{
name: "Get an Existing Worker",
scopeId: worker.GetScopeId(),
req: &pbs.GetWorkerRequest{Id: worker.GetPublicId()},
res: &pbs.GetWorkerResponse{Item: wantWorker},
},
{
name: "Get a non existant Worker",
req: &pbs.GetWorkerRequest{Id: servers.WorkerPrefix + "_DoesntExis"},
res: nil,
err: handlers.ApiErrorWithCode(codes.NotFound),
},
{
name: "Wrong id prefix",
req: &pbs.GetWorkerRequest{Id: "j_1234567890"},
res: nil,
err: handlers.ApiErrorWithCode(codes.InvalidArgument),
},
{
name: "space in id",
req: &pbs.GetWorkerRequest{Id: servers.WorkerPrefix + "_1 23456789"},
res: nil,
err: handlers.ApiErrorWithCode(codes.InvalidArgument),
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
s, err := NewService(context.Background(), repoFn, iamRepoFn)
require.NoError(t, err, "Couldn't create new worker service.")
got, err := s.GetWorker(auth.DisabledAuthTestContext(iamRepoFn, tc.scopeId), tc.req)
if tc.err != nil {
require.Error(t, err)
assert.True(t, errors.Is(err, tc.err), "GetWorker(%+v) got error %v, wanted %v", tc.req, err, tc.err)
} else {
require.NoError(t, err)
}
assert.Empty(t, cmp.Diff(got, tc.res, protocmp.Transform()), "GetWorker(%q) got response\n%q, wanted\n%q", tc.req, got, tc.res)
})
}
}
func TestList(t *testing.T) {
conn, _ := db.TestSetup(t, "postgres")
wrap := db.TestWrapper(t)
iamRepo := iam.TestRepo(t, conn, wrap)
iamRepoFn := func() (*iam.Repository, error) {
return iamRepo, nil
}
rw := db.New(conn)
kms := kms.TestKms(t, conn, wrap)
repoFn := func() (*servers.Repository, error) {
return servers.NewRepository(rw, rw, kms)
}
var wantWorkers []*pb.Worker
for i := 0; i < 10; i++ {
w := servers.TestWorker(t, conn, wrap, servers.WithName(fmt.Sprintf("worker%d", i)))
wantWorkers = append(wantWorkers, &pb.Worker{
Id: w.GetPublicId(),
ScopeId: w.GetScopeId(),
Scope: &scopes.ScopeInfo{Id: w.GetScopeId(), Type: scope.Global.String(), Name: scope.Global.String(), Description: "Global Scope"},
CreatedTime: w.CreateTime.GetTimestamp(),
UpdatedTime: w.UpdateTime.GetTimestamp(),
Version: w.GetVersion(),
Name: wrapperspb.String(w.GetName()),
AuthorizedActions: testAuthorizedActions,
CanonicalAddress: w.CanonicalAddress(),
LastStatusTime: w.GetLastStatusTime().GetTimestamp(),
WorkerConfig: &pb.WorkerConfig{
Address: w.GetWorkerReportedAddress(),
Name: w.GetWorkerReportedName(),
},
})
}
cases := []struct {
name string
req *pbs.ListWorkersRequest
res *pbs.ListWorkersResponse
err error
}{
{
name: "List All Workers",
req: &pbs.ListWorkersRequest{ScopeId: scope.Global.String()},
res: &pbs.ListWorkersResponse{Items: wantWorkers},
},
{
name: "List global workers recursively",
req: &pbs.ListWorkersRequest{ScopeId: "global", Recursive: true},
res: &pbs.ListWorkersResponse{
Items: wantWorkers,
},
},
{
name: "Filter to a single workers",
req: &pbs.ListWorkersRequest{ScopeId: "global", Recursive: true, Filter: `"/item/name"=="worker2"`},
res: &pbs.ListWorkersResponse{
Items: wantWorkers[2:3],
},
},
{
name: "Filter to 2 workers",
req: &pbs.ListWorkersRequest{ScopeId: "global", Recursive: true, Filter: `"/item/name" matches "worker[23]"`},
res: &pbs.ListWorkersResponse{
Items: wantWorkers[2:4],
},
},
{
name: "Filter to no workers",
req: &pbs.ListWorkersRequest{ScopeId: "global", Recursive: true, Filter: `"/item/id"=="doesntmatch"`},
res: &pbs.ListWorkersResponse{},
},
{
name: "Filter Bad Format",
req: &pbs.ListWorkersRequest{ScopeId: "global", Filter: `"//id/"=="bad"`},
err: handlers.InvalidArgumentErrorf("bad format", nil),
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
assert, require := assert.New(t), require.New(t)
s, err := NewService(context.Background(), repoFn, iamRepoFn)
require.NoError(err, "Couldn't create new worker service.")
// Test with a non-anon user
got, gErr := s.ListWorkers(auth.DisabledAuthTestContext(iamRepoFn, tc.req.GetScopeId()), tc.req)
if tc.err != nil {
require.Error(gErr)
assert.True(errors.Is(gErr, tc.err), "ListWorkers(%q) got error %v, wanted %v", tc.req.GetScopeId(), gErr, tc.err)
return
}
require.NoError(gErr)
sort.Slice(got.Items, func(i, j int) bool {
return got.Items[i].GetName().GetValue() < got.Items[j].GetName().GetValue()
})
assert.Empty(cmp.Diff(got, tc.res, protocmp.Transform()), "ListWorkers(%q) got response %q, wanted %q", tc.req.GetScopeId(), got, tc.res)
// Test the anon case
got, gErr = s.ListWorkers(auth.DisabledAuthTestContext(iamRepoFn, tc.req.GetScopeId(), auth.WithUserId(auth.AnonymousUserId)), tc.req)
require.NoError(gErr)
assert.Len(got.Items, len(tc.res.Items))
for _, item := range got.GetItems() {
require.Nil(item.CreatedTime)
require.Nil(item.UpdatedTime)
require.Zero(item.Version)
}
})
}
}