diff --git a/.drone.env b/.drone.env index 9139a1ae977..df1245bc2f6 100644 --- a/.drone.env +++ b/.drone.env @@ -1,4 +1,4 @@ # The test runner source for API tests -APITESTS_COMMITID=9ac0452d61f062572f7e4663679ffb8ac06845e6 +APITESTS_COMMITID=00aeefac3b23efbb4b7fe67225e74a8a6e6b71f9 APITESTS_BRANCH=master APITESTS_REPO_GIT_URL=https://github.com/owncloud/ocis.git diff --git a/changelog/unreleased/fix-publicshare-list-stat-fanout.md b/changelog/unreleased/fix-publicshare-list-stat-fanout.md new file mode 100644 index 00000000000..c6e377ba9ca --- /dev/null +++ b/changelog/unreleased/fix-publicshare-list-stat-fanout.md @@ -0,0 +1,14 @@ +Bugfix: Stop statting every public link when listing shares unfiltered + +ListPublicShares issued one gateway Stat per public link not created by the +calling user in order to check the ListGrants permission. On tenants with a few +thousand links this exceeded the request deadline and the shares list returned +nothing. The permission check itself is unchanged, so every user sees exactly +the same links as before. Instead the manager now stats each distinct resource +at most once per request, caching allowed and denied answers alike, and runs +those stats through a bounded worker pool rather than one after another. When +the caller's own deadline is about to expire the remaining stats are abandoned +and a partial list is returned with a warning, instead of the whole request +failing. + +https://github.com/owncloud/reva/pull/712 diff --git a/pkg/publicshare/manager/json/json.go b/pkg/publicshare/manager/json/json.go index 7a998ce1e9f..a72e41b73ab 100644 --- a/pkg/publicshare/manager/json/json.go +++ b/pkg/publicshare/manager/json/json.go @@ -25,20 +25,22 @@ import ( "os" "os/signal" "strconv" - "strings" "sync" "syscall" "time" "github.com/rs/zerolog/log" "golang.org/x/crypto/bcrypt" + "golang.org/x/sync/errgroup" "google.golang.org/protobuf/proto" + gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1" user "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1" link "github.com/cs3org/go-cs3apis/cs3/sharing/link/v1beta1" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" typespb "github.com/cs3org/go-cs3apis/cs3/types/v1beta1" + "github.com/mitchellh/mapstructure" "github.com/owncloud/reva/v2/pkg/appctx" "github.com/owncloud/reva/v2/pkg/errtypes" "github.com/owncloud/reva/v2/pkg/publicshare" @@ -49,9 +51,10 @@ import ( "github.com/owncloud/reva/v2/pkg/publicshare/manager/registry" "github.com/owncloud/reva/v2/pkg/rgrpc/todo/pool" "github.com/owncloud/reva/v2/pkg/storage/utils/metadata" + "github.com/owncloud/reva/v2/pkg/storagespace" "github.com/owncloud/reva/v2/pkg/utils" - "github.com/mitchellh/mapstructure" "github.com/pkg/errors" + "google.golang.org/protobuf/types/known/fieldmaskpb" ) func init() { @@ -107,6 +110,15 @@ func NewCS3(c map[string]interface{}) (publicshare.Manager, error) { return New(conf.GatewayAddr, conf.SharePasswordHashCost, conf.JanitorRunInterval, conf.EnableExpiredSharesCleanup, p) } +// defaultStatConcurrency bounds how many Stat RPCs ListPublicShares may have +// in flight at once while checking ListGrants on distinct foreign resources. +// 5 mirrors the default concurrency the decomposedfs share manager clamps to +// for the same kind of bounded fan-out (see +// pkg/storage/utils/decomposedfs/options/options.go): enough to make a dent +// in a large batch of distinct resources without opening so many concurrent +// Stat RPCs that the gateway itself becomes the bottleneck. +const defaultStatConcurrency = 5 + // New returns a new public share manager instance func New(gwAddr string, pwHashCost, janitorRunInterval int, enableCleanup bool, p persistence.Persistence) (publicshare.Manager, error) { m := &manager{ @@ -116,6 +128,7 @@ func New(gwAddr string, pwHashCost, janitorRunInterval int, enableCleanup bool, janitorRunInterval: janitorRunInterval, enableExpiredSharesCleanup: enableCleanup, persistence: p, + maxConcurrency: defaultStatConcurrency, } go m.startJanitorRun() @@ -161,6 +174,10 @@ type manager struct { passwordHashCost int janitorRunInterval int enableExpiredSharesCleanup bool + + // maxConcurrency bounds how many Stat RPCs ListPublicShares may have in + // flight at once while checking ListGrants on distinct foreign resources. + maxConcurrency int } func (m *manager) init() error { @@ -482,6 +499,23 @@ func (m *manager) GetPublicShare(ctx context.Context, u *user.User, ref *link.Pu } // ListPublicShares retrieves all the shares on the manager that are valid. +// +// Visibility of a foreign share (one not created by the calling user) is +// decided by a per-resource Stat, exactly as it always was: ListGrants on the +// share's resource is the OR of every ACE from that resource up to the space +// root (see assemblePermissions in +// pkg/storage/utils/decomposedfs/node/permissions.go, which also +// short-circuits on deny grants), so it cannot be derived from any +// precomputed set of space or resource ids without risking either false +// negatives or a privilege escalation (OCISDEV-861). What this method bounds +// is the *cost* of that check: pass 1 collects the set of distinct resources +// referenced by foreign shares, and pass 2 stats each of them at most once, +// concurrently, within a time budget derived from the caller's own context +// deadline (see statBudgetContext). N links on M distinct resources thus +// costs at most M stats, not N, and the whole call is bounded by whatever +// deadline the caller supplied, regardless of how large M is. If the caller +// supplies no deadline, the stat fan-out is unbounded, matching pre-existing +// behaviour. func (m *manager) ListPublicShares(ctx context.Context, u *user.User, filters []*link.ListPublicSharesRequest_Filter, sign bool) ([]*link.PublicShare, error) { m.mutex.Lock() defer m.mutex.Unlock() @@ -497,13 +531,17 @@ func (m *manager) ListPublicShares(ctx context.Context, u *user.User, filters [] return nil, err } - client, err := pool.GetGatewayServiceClient(m.gatewayAddr) - if err != nil { - return nil, errors.Wrap(err, "failed to list shares") - } - cache := make(map[string]struct{}) + // Pass 1 (in-memory, no RPCs): decode every persisted share once, handle + // expiry and filters exactly as before, and split the survivors into + // shares the caller created (no permission check needed) and foreign + // shares (which do need one). While doing so, collect the set of + // distinct resources the foreign shares point at, keyed by + // storagespace.FormatResourceID, so pass 2 can stat each of them exactly + // once. + ownShares := make([]*publicShare, 0) + foreignShares := make([]*publicShare, 0) + foreignResourceIDs := make(map[string]*provider.ResourceId) - shares := []*link.PublicShare{} for _, v := range db { var local publicShare if err := utils.UnmarshalJSONToProtoV1([]byte(v.(map[string]interface{})["share"].(string)), &local.PublicShare); err != nil { @@ -531,51 +569,221 @@ func (m *manager) ListPublicShares(ctx context.Context, u *user.User, filters [] continue } - key := strings.Join([]string{local.ResourceId.StorageId, local.ResourceId.OpaqueId}, "!") - if _, hit := cache[key]; !hit && !publicshare.IsCreatedByUser(&local.PublicShare, u) { - sRes, err := client.Stat(ctx, &provider.StatRequest{Ref: &provider.Reference{ResourceId: local.ResourceId}}) - if err != nil { - log.Error(). - Err(err). - Interface("resource_id", local.ResourceId). - Msg("ListShares: an error occurred during stat on the resource") - continue - } - if sRes.Status.Code != rpc.Code_CODE_OK { - if sRes.Status.Code == rpc.Code_CODE_NOT_FOUND { - log.Debug(). - Str("message", sRes.Status.Message). - Interface("status", sRes.Status). - Interface("resource_id", local.ResourceId). - Msg("ListShares: Resource not found") - continue - } - log.Error(). - Str("message", sRes.Status.Message). - Interface("status", sRes.Status). - Interface("resource_id", local.ResourceId). - Msg("ListShares: could not stat resource") - continue - } - if !sRes.Info.PermissionSet.ListGrants { - // skip because the user doesn't have the permissions to list - // shares of this file. - continue - } - cache[key] = struct{}{} + if publicshare.IsCreatedByUser(&local.PublicShare, u) { + ownShares = append(ownShares, &local) + continue } + foreignShares = append(foreignShares, &local) + foreignResourceIDs[storagespace.FormatResourceID(local.ResourceId)] = local.ResourceId + } + + // Pass 2 (bounded RPCs): stat each distinct foreign resource once, + // concurrently, within a time budget. A caller who created every share + // (or has no foreign shares surviving the filters) issues no RPC at all. + var permitted map[string]bool + if len(foreignResourceIDs) > 0 { + client, err := pool.GetGatewayServiceClient(m.gatewayAddr) + if err != nil { + return nil, errors.Wrap(err, "failed to list shares") + } + permitted = m.statForeignResources(ctx, u, client, foreignResourceIDs) + } + + shares := make([]*link.PublicShare, 0, len(ownShares)+len(foreignShares)) + for _, local := range ownShares { + if local.PublicShare.PasswordProtected && sign { + if err := publicshare.AddSignature(&local.PublicShare, local.Password); err != nil { + return nil, err + } + } + shares = append(shares, &local.PublicShare) + } + for _, local := range foreignShares { + // Any resource whose permission was never determined (e.g. because + // the time budget ran out) is absent here and therefore excluded: + // fail closed, never include a share whose permission is unknown. + if !permitted[storagespace.FormatResourceID(local.ResourceId)] { + continue + } if local.PublicShare.PasswordProtected && sign { if err := publicshare.AddSignature(&local.PublicShare, local.Password); err != nil { return nil, err } } - shares = append(shares, &local.PublicShare) } return shares, nil } +// statForeignResources stats each of the given distinct resources at most +// once, using a bounded pool of at most m.maxConcurrency concurrent workers, +// and returns a map of storagespace.FormatResourceID -> whether the calling +// user may list grants on that resource. +// +// The whole operation is bounded by a time budget derived from the caller's +// own context deadline, minus a small margin (see statBudgetContext): if the +// budget runs out before every resource has been stated, statting stops, a +// single warning is logged naming how many resources were skipped, and the +// partial map is returned rather than letting the caller block until its own +// deadline cancels the whole request with code = Canceled (OCISDEV-861). If +// the caller supplies no deadline, no budget is imposed. Any resource not +// present in the returned map was never decided and must be treated as not +// permitted by the caller. +func (m *manager) statForeignResources(ctx context.Context, u *user.User, client gateway.GatewayAPIClient, resourceIDs map[string]*provider.ResourceId) map[string]bool { + log := appctx.GetLogger(ctx) + + statCtx, cancel := m.statBudgetContext(ctx) + defer cancel() + + results := newStatResults() + + numWorkers := m.maxConcurrency + if numWorkers > len(resourceIDs) { + numWorkers = len(resourceIDs) + } + if numWorkers < 1 { + numWorkers = 1 + } + + type job struct { + rid *provider.ResourceId + } + jobs := make(chan job) + + g, gctx := errgroup.WithContext(statCtx) + + // Distribute work. Stop feeding jobs once the budget runs out so workers + // drain and exit instead of blocking forever on a full channel. + g.Go(func() error { + defer close(jobs) + for _, rid := range resourceIDs { + select { + case jobs <- job{rid}: + case <-gctx.Done(): + return nil + } + } + return nil + }) + + // Spawn workers that concurrently work the queue, bounded by + // numWorkers <= m.maxConcurrency concurrent Stat RPCs in flight. + for i := 0; i < numWorkers; i++ { + g.Go(func() error { + for j := range jobs { + if gctx.Err() != nil { + // Budget exhausted: stop statting. A resource left + // undecided is simply absent from the returned map, so + // the caller treats it as not permitted. + continue + } + m.userCanListGrants(statCtx, client, results, j.rid) + } + return nil + }) + } + _ = g.Wait() + + result := results.snapshot() + if skipped := len(resourceIDs) - len(result); skipped > 0 { + log.Warn(). + Str("user_id", u.GetId().GetOpaqueId()). + Int("resources_checked", len(result)). + Int("resources_skipped", skipped). + Msg("ListPublicShares: stat time budget exhausted before every resource could be checked, returned list may be incomplete") + } + return result +} + +// statBudgetContext derives a child context bounding how long +// statForeignResources may spend statting resources. The budget is derived +// solely from the caller's own context deadline, minus a small margin so the +// rest of the request (decoding, filtering, signing) still has time to run +// before the caller's deadline fires: this method never imposes a bound of +// its own. If the incoming context has no deadline, the returned context has +// none either, and the stat fan-out is unbounded - the same as before this +// budget existed. +func (m *manager) statBudgetContext(ctx context.Context) (context.Context, context.CancelFunc) { + const margin = 200 * time.Millisecond + + deadline, ok := ctx.Deadline() + if !ok { + return ctx, func() {} + } + + budget := time.Until(deadline) - margin + if budget < 0 { + budget = 0 + } + return context.WithTimeout(ctx, budget) +} + +// statResults collects ListGrants answers for resources, keyed by +// storagespace.FormatResourceID. It is safe for concurrent use by the bounded +// worker pool in statForeignResources. +type statResults struct { + mu sync.Mutex + data map[string]bool +} + +func newStatResults() *statResults { + return &statResults{data: make(map[string]bool)} +} + +func (r *statResults) set(key string, allowed bool) { + r.mu.Lock() + defer r.mu.Unlock() + r.data[key] = allowed +} + +// snapshot returns a copy of the results collected so far. Call only once no +// more writers are running (e.g. after an errgroup.Wait), or take a copy +// under the same lock discipline as set. +func (r *statResults) snapshot() map[string]bool { + r.mu.Lock() + defer r.mu.Unlock() + out := make(map[string]bool, len(r.data)) + for k, v := range r.data { + out[k] = v + } + return out +} + +// userCanListGrants reports whether the current user may list grants on the +// given resource and records the answer in results. The resource IDs are +// already deduplicated by the caller (see the foreignResourceIDs map built +// in ListPublicShares) before statForeignResources ever runs, so each +// resource is stated at most once and there is nothing to look up here +// beforehand. +func (m *manager) userCanListGrants(ctx context.Context, client gateway.GatewayAPIClient, results *statResults, rid *provider.ResourceId) bool { + log := appctx.GetLogger(ctx) + key := storagespace.FormatResourceID(rid) + + sRes, err := client.Stat(ctx, &provider.StatRequest{ + Ref: &provider.Reference{ResourceId: rid}, + FieldMask: &fieldmaskpb.FieldMask{Paths: []string{"permissions"}}, + }) + switch { + case err != nil: + log.Error().Err(err).Interface("resource_id", rid).Msg("ListShares: an error occurred during stat on the resource") + results.set(key, false) + return false + case sRes.Status.Code == rpc.Code_CODE_NOT_FOUND: + log.Debug().Str("message", sRes.Status.Message).Interface("status", sRes.Status).Interface("resource_id", rid).Msg("ListShares: Resource not found") + results.set(key, false) + return false + case sRes.Status.Code != rpc.Code_CODE_OK: + log.Error().Str("message", sRes.Status.Message).Interface("status", sRes.Status).Interface("resource_id", rid).Msg("ListShares: could not stat resource") + results.set(key, false) + return false + } + + allowed := sRes.GetInfo().GetPermissionSet().GetListGrants() + results.set(key, allowed) + return allowed +} + func (m *manager) cleanupExpiredShares() { m.mutex.Lock() defer m.mutex.Unlock() diff --git a/pkg/publicshare/manager/json/json_export_test.go b/pkg/publicshare/manager/json/json_export_test.go new file mode 100644 index 00000000000..7762606c146 --- /dev/null +++ b/pkg/publicshare/manager/json/json_export_test.go @@ -0,0 +1,33 @@ +// Copyright 2018-2021 CERN +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// In applying this license, CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +package json + +import ( + "github.com/owncloud/reva/v2/pkg/publicshare" +) + +// SetStatConcurrency overrides the bounded worker pool size used by +// ListPublicShares' stat fan-out. maxConcurrency is deliberately not +// operator-configurable (see defaultStatConcurrency), but tests still need to +// exercise the bound directly rather than waiting out the real default. +// +// Export for testing only. +func SetStatConcurrency(m publicshare.Manager, n int) { + m.(*manager).maxConcurrency = n +} diff --git a/pkg/publicshare/manager/json/json_test.go b/pkg/publicshare/manager/json/json_test.go index f6036fd14a6..b595960e54c 100644 --- a/pkg/publicshare/manager/json/json_test.go +++ b/pkg/publicshare/manager/json/json_test.go @@ -21,11 +21,14 @@ package json_test import ( "context" encjson "encoding/json" + "errors" + "fmt" "os" "path/filepath" "sync" "time" + gatewayv1beta1 "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1" userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" link "github.com/cs3org/go-cs3apis/cs3/sharing/link/v1beta1" providerv1beta1 "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" @@ -33,8 +36,14 @@ import ( "github.com/owncloud/reva/v2/pkg/publicshare" "github.com/owncloud/reva/v2/pkg/publicshare/manager/json" "github.com/owncloud/reva/v2/pkg/publicshare/manager/json/persistence/cs3" + "github.com/owncloud/reva/v2/pkg/rgrpc/status" + "github.com/owncloud/reva/v2/pkg/rgrpc/todo/pool" "github.com/owncloud/reva/v2/pkg/storage/utils/metadata" + "github.com/owncloud/reva/v2/pkg/utils" + "github.com/owncloud/reva/v2/tests/cs3mocks/mocks" + "github.com/stretchr/testify/mock" "golang.org/x/crypto/bcrypt" + "google.golang.org/grpc" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/testing/protocmp" @@ -73,6 +82,7 @@ var _ = Describe("Json", func() { m publicshare.Manager tmpFile *os.File ctx context.Context + client *mocks.GatewayAPIClient ) Context("with a file persistence layer", func() { @@ -86,6 +96,15 @@ var _ = Describe("Json", func() { "file": tmpFile.Name(), "gateway_addr": "https://localhost:9200", } + + pool.RemoveSelector("GatewaySelector" + "https://localhost:9200") + client = &mocks.GatewayAPIClient{} + pool.GetSelector[gatewayv1beta1.GatewayAPIClient]( + "GatewaySelector", + "https://localhost:9200", + func(cc grpc.ClientConnInterface) gatewayv1beta1.GatewayAPIClient { return client }, + ) + m, err = json.NewFile(config) Expect(err).ToNot(HaveOccurred()) @@ -132,6 +151,45 @@ var _ = Describe("Json", func() { }) Describe("ListPublicShares", func() { + type shareSpec struct { + ID, Token string + Creator *userpb.UserId + StorageID, SpaceID, OpaqueID string + } + + seedShares := func(path string, specs []shareSpec) { + raw, err := os.ReadFile(path) + ExpectWithOffset(1, err).ToNot(HaveOccurred()) + db := map[string]interface{}{} + if len(raw) > 0 { + ExpectWithOffset(1, encjson.Unmarshal(raw, &db)).To(Succeed()) + } + + for _, s := range specs { + ps := &link.PublicShare{ + Id: &link.PublicShareId{OpaqueId: s.ID}, + Token: s.Token, + Creator: s.Creator, + Owner: s.Creator, + ResourceId: &providerv1beta1.ResourceId{StorageId: s.StorageID, SpaceId: s.SpaceID, OpaqueId: s.OpaqueID}, + } + enc, err := utils.MarshalProtoV1ToJSON(ps) + ExpectWithOffset(1, err).ToNot(HaveOccurred()) + db[s.ID] = map[string]interface{}{"share": string(enc), "password": ""} + } + patched, err := encjson.Marshal(db) + ExpectWithOffset(1, err).ToNot(HaveOccurred()) + ExpectWithOffset(1, os.WriteFile(path, patched, 0644)).To(Succeed()) + } + + opaqueIDs := func(shares []*link.PublicShare) []string { + out := make([]string, 0, len(shares)) + for _, s := range shares { + out = append(out, s.Id.OpaqueId) + } + return out + } + It("skips shares whose persisted resource_id is nil instead of panicking", func() { // Create one valid share so the manager has a healthy row to compare against. validShare, err := m.CreatePublicShare(ctx, user1, sharedResource, &link.Grant{ @@ -164,6 +222,406 @@ var _ = Describe("Json", func() { Expect(len(shares)).To(Equal(1)) Expect(shares[0].Id.OpaqueId).To(Equal(validShare.Id.OpaqueId)) }) + + // Headline correctness property (OCISDEV-861): the returned set must be + // exactly own shares plus foreign shares the caller may list grants on, + // per the *unchanged* per-resource Stat check. A resource-narrowing + // filter that still matches every seeded share must not change that: + // there is no more special-casing between filtered and unfiltered + // requests, both go through the very same statForeignResources path. + It("returns exactly own shares plus permitted foreign shares, identically whether or not a resource filter is passed", func() { + user2 := &userpb.UserId{Idp: "https://localhost:9200", OpaqueId: "einstein"} + ridPermitted := &providerv1beta1.ResourceId{StorageId: "storageid", SpaceId: "space-a", OpaqueId: "oa"} + ridDenied := &providerv1beta1.ResourceId{StorageId: "storageid", SpaceId: "space-b", OpaqueId: "ob"} + seedShares(tmpFile.Name(), []shareSpec{ + {ID: "own-1", Token: "t-own-1", Creator: user1.Id, StorageID: "storageid", SpaceID: "space-own", OpaqueID: "oown"}, + {ID: "foreign-permitted", Token: "t-fp", Creator: user2, StorageID: ridPermitted.StorageId, SpaceID: ridPermitted.SpaceId, OpaqueID: ridPermitted.OpaqueId}, + {ID: "foreign-denied", Token: "t-fd", Creator: user2, StorageID: ridDenied.StorageId, SpaceID: ridDenied.SpaceId, OpaqueID: ridDenied.OpaqueId}, + }) + + client.On("Stat", mock.Anything, mock.MatchedBy(func(req *providerv1beta1.StatRequest) bool { + return utils.ResourceIDEqual(req.GetRef().GetResourceId(), ridPermitted) + })).Return(&providerv1beta1.StatResponse{ + Status: status.NewOK(ctx), + Info: &providerv1beta1.ResourceInfo{PermissionSet: &providerv1beta1.ResourcePermissions{ListGrants: true}}, + }, nil) + client.On("Stat", mock.Anything, mock.MatchedBy(func(req *providerv1beta1.StatRequest) bool { + return utils.ResourceIDEqual(req.GetRef().GetResourceId(), ridDenied) + })).Return(&providerv1beta1.StatResponse{ + Status: status.NewOK(ctx), + Info: &providerv1beta1.ResourceInfo{PermissionSet: &providerv1beta1.ResourcePermissions{ListGrants: false}}, + }, nil) + + unfiltered, err := m.ListPublicShares(ctx, user1, nil, false) + Expect(err).ToNot(HaveOccurred()) + Expect(opaqueIDs(unfiltered)).To(ConsistOf("own-1", "foreign-permitted")) + + // Every seeded share lives under storage "storageid", so this filter + // narrows nothing away; it exists purely to prove filtered and + // unfiltered requests are decided identically. + filtered, err := m.ListPublicShares(ctx, user1, []*link.ListPublicSharesRequest_Filter{ + publicshare.StorageIDFilter("storageid"), + }, false) + Expect(err).ToNot(HaveOccurred()) + Expect(opaqueIDs(filtered)).To(ConsistOf("own-1", "foreign-permitted")) + }) + + It("stats each distinct resource at most once regardless of how many links point at it", func() { + user2 := &userpb.UserId{Idp: "https://localhost:9200", OpaqueId: "einstein"} + const numResources = 5 + const numLinks = 500 + specs := make([]shareSpec, 0, numLinks) + for i := 0; i < numLinks; i++ { + specs = append(specs, shareSpec{ + ID: fmt.Sprintf("foreign-%d", i), + Token: fmt.Sprintf("t-foreign-%d", i), + Creator: user2, + StorageID: "storageid", + SpaceID: "space-foreign", + OpaqueID: fmt.Sprintf("o-%d", i%numResources), + }) + } + seedShares(tmpFile.Name(), specs) + + client.On("Stat", mock.Anything, mock.Anything).Return(&providerv1beta1.StatResponse{ + Status: status.NewOK(ctx), + Info: &providerv1beta1.ResourceInfo{PermissionSet: &providerv1beta1.ResourcePermissions{ListGrants: true}}, + }, nil) + + shares, err := m.ListPublicShares(ctx, user1, nil, false) + Expect(err).ToNot(HaveOccurred()) + Expect(shares).To(HaveLen(numLinks)) + // The point of the ticket: N links on M distinct resources costs M + // stats, not N. + client.AssertNumberOfCalls(GinkgoT(), "Stat", numResources) + }) + + It("bounds the number of concurrent Stat calls to maxConcurrency", func() { + user2 := &userpb.UserId{Idp: "https://localhost:9200", OpaqueId: "einstein"} + const numResources = 10 + specs := make([]shareSpec, 0, numResources) + for i := 0; i < numResources; i++ { + specs = append(specs, shareSpec{ + ID: fmt.Sprintf("foreign-%d", i), + Token: fmt.Sprintf("t-foreign-%d", i), + Creator: user2, + StorageID: "storageid", + SpaceID: "space-foreign", + OpaqueID: fmt.Sprintf("o-%d", i), + }) + } + seedShares(tmpFile.Name(), specs) + + var mu sync.Mutex + var inFlight, maxInFlight int + client.On("Stat", mock.Anything, mock.Anything).Run(func(_ mock.Arguments) { + mu.Lock() + inFlight++ + if inFlight > maxInFlight { + maxInFlight = inFlight + } + mu.Unlock() + + time.Sleep(20 * time.Millisecond) + + mu.Lock() + inFlight-- + mu.Unlock() + }).Return(&providerv1beta1.StatResponse{ + Status: status.NewOK(ctx), + Info: &providerv1beta1.ResourceInfo{PermissionSet: &providerv1beta1.ResourcePermissions{ListGrants: true}}, + }, nil) + + bounded, err := json.NewFile(map[string]interface{}{ + "file": tmpFile.Name(), + "gateway_addr": "https://localhost:9200", + }) + Expect(err).ToNot(HaveOccurred()) + json.SetStatConcurrency(bounded, 2) + + shares, err := bounded.ListPublicShares(ctx, user1, nil, false) + Expect(err).ToNot(HaveOccurred()) + Expect(shares).To(HaveLen(numResources)) + + mu.Lock() + defer mu.Unlock() + // Never more than maxConcurrency Stat calls in flight at once... + Expect(maxInFlight).To(BeNumerically("<=", 2)) + // ...but concurrency is actually happening, not accidentally serial. + Expect(maxInFlight).To(BeNumerically(">=", 2)) + }) + + It("fails closed when the caller's deadline leaves no budget at all", func() { + user2 := &userpb.UserId{Idp: "https://localhost:9200", OpaqueId: "einstein"} + seedShares(tmpFile.Name(), []shareSpec{ + {ID: "own-1", Token: "t-own-1", Creator: user1.Id, StorageID: "storageid", SpaceID: "space-own", OpaqueID: "oown"}, + {ID: "foreign-1", Token: "t-f1", Creator: user2, StorageID: "storageid", SpaceID: "space-x", OpaqueID: "o1"}, + }) + + // A Stat that takes far longer than the caller's deadline, but still + // honours context cancellation the way a real gRPC client would. It + // must never actually be invoked below: see the assertion at the end. + client.On("Stat", mock.Anything, mock.Anything).Return( + func(ctx context.Context, _ *providerv1beta1.StatRequest, _ ...grpc.CallOption) (*providerv1beta1.StatResponse, error) { + select { + case <-time.After(50 * time.Millisecond): + return &providerv1beta1.StatResponse{ + Status: status.NewOK(ctx), + Info: &providerv1beta1.ResourceInfo{PermissionSet: &providerv1beta1.ResourcePermissions{ListGrants: true}}, + }, nil + case <-ctx.Done(): + return nil, ctx.Err() + } + }) + + // statBudgetContext computes time.Until(deadline) - margin, and the + // 200ms margin alone dwarfs this 5ms deadline, so the derived budget + // clamps to zero: the fan-out starts with an already-expired context, + // and every worker takes the gctx.Err() != nil branch on its very + // first job, before ever calling Stat. This is the "no budget at all" + // case; see the spec below for a genuine mid-flight cancellation. + shortCtx, cancel := context.WithTimeout(ctx, 5*time.Millisecond) + defer cancel() + + start := time.Now() + shares, err := m.ListPublicShares(shortCtx, user1, nil, false) + elapsed := time.Since(start) + + // Partial list beats code = Canceled: no error... + Expect(err).ToNot(HaveOccurred()) + // ...own shares are unaffected by the budget... + Expect(opaqueIDs(shares)).To(ConsistOf("own-1")) + // ...and a resource whose permission was never determined is never + // included (fail closed). + Expect(opaqueIDs(shares)).ToNot(ContainElement("foreign-1")) + // The call must not block for the full 50ms Stat delay. + Expect(elapsed).To(BeNumerically("<", 40*time.Millisecond)) + // The budget was already exhausted on entry, so Stat is never called. + client.AssertNumberOfCalls(GinkgoT(), "Stat", 0) + }) + + It("fails closed and returns a partial list when the caller's own deadline expires mid-flight", func() { + user2 := &userpb.UserId{Idp: "https://localhost:9200", OpaqueId: "einstein"} + ridA := &providerv1beta1.ResourceId{StorageId: "storageid", SpaceId: "space-x", OpaqueId: "o1"} + ridB := &providerv1beta1.ResourceId{StorageId: "storageid", SpaceId: "space-x", OpaqueId: "o2"} + ridC := &providerv1beta1.ResourceId{StorageId: "storageid", SpaceId: "space-x", OpaqueId: "o3"} + seedShares(tmpFile.Name(), []shareSpec{ + {ID: "own-1", Token: "t-own-1", Creator: user1.Id, StorageID: "storageid", SpaceID: "space-own", OpaqueID: "oown"}, + {ID: "foreign-a", Token: "t-fa", Creator: user2, StorageID: ridA.StorageId, SpaceID: ridA.SpaceId, OpaqueID: ridA.OpaqueId}, + {ID: "foreign-b", Token: "t-fb", Creator: user2, StorageID: ridB.StorageId, SpaceID: ridB.SpaceId, OpaqueID: ridB.OpaqueId}, + {ID: "foreign-c", Token: "t-fc", Creator: user2, StorageID: ridC.StorageId, SpaceID: ridC.SpaceId, OpaqueID: ridC.OpaqueId}, + }) + + // A single worker, so the three distinct resources are stated strictly + // one after another rather than all at once. Each Stat costs ~120ms + // against a residual budget of ~200ms (see below), so the first + // resource is decided and the rest are cut off. That asymmetry is the + // point: this spec pins a genuinely PARTIAL result, where the "no + // budget at all" spec above pins the degenerate case in which nothing + // is decided. + // + // Note on what enforces this: cancellation is honoured by the Stat + // call itself (a real gRPC client aborts on a cancelled context, as + // the mock below does). The worker's own gctx check is an + // optimisation that avoids dispatching doomed RPCs, so removing it + // does not change the outcome here. The property that actually keeps + // this fail-closed is that an undecided resource is absent from the + // permitted map — pinned by the "excludes a foreign share when Stat + // fails closed" table below. + json.SetStatConcurrency(m, 1) + + client.On("Stat", mock.Anything, mock.Anything).Return( + func(ctx context.Context, _ *providerv1beta1.StatRequest, _ ...grpc.CallOption) (*providerv1beta1.StatResponse, error) { + select { + case <-time.After(120 * time.Millisecond): + return &providerv1beta1.StatResponse{ + Status: status.NewOK(ctx), + Info: &providerv1beta1.ResourceInfo{PermissionSet: &providerv1beta1.ResourcePermissions{ListGrants: true}}, + }, nil + case <-ctx.Done(): + return nil, ctx.Err() + } + }) + + // Residual budget is roughly 400ms - 200ms margin = ~200ms, so one or + // two of the ~120ms Stats complete and the remainder are abandoned. + midCtx, cancel := context.WithTimeout(ctx, 400*time.Millisecond) + defer cancel() + + start := time.Now() + shares, err := m.ListPublicShares(midCtx, user1, nil, false) + elapsed := time.Since(start) + returned := opaqueIDs(shares) + + // Partial list beats code = Canceled: no error... + Expect(err).ToNot(HaveOccurred()) + // ...own shares are never subject to the budget... + Expect(returned).To(ContainElement("own-1")) + // ...at least one foreign share WAS decided permitted before the + // budget ran out, so this is a partial result and not an empty one... + foreign := 0 + for _, id := range returned { + if id != "own-1" { + foreign++ + } + } + Expect(foreign).To(BeNumerically(">", 0), "expected at least one foreign share to be decided before the budget expired") + // ...but not all of them: the remainder were abandoned undecided and + // are therefore excluded (fail closed, never fail open). + Expect(foreign).To(BeNumerically("<", 3), "expected the budget to cut the stat fan-out short") + // The call must not block for all three Stat delays. + Expect(elapsed).To(BeNumerically("<", 360*time.Millisecond)) + }) + + It("does not truncate the list when the caller grants a generous deadline", func() { + user2 := &userpb.UserId{Idp: "https://localhost:9200", OpaqueId: "einstein"} + const numResources = 5 + specs := make([]shareSpec, 0, numResources) + for i := 0; i < numResources; i++ { + specs = append(specs, shareSpec{ + ID: fmt.Sprintf("foreign-%d", i), + Token: fmt.Sprintf("t-foreign-%d", i), + Creator: user2, + StorageID: "storageid", + SpaceID: "space-foreign", + OpaqueID: fmt.Sprintf("o-%d", i), + }) + } + seedShares(tmpFile.Name(), specs) + + // Each Stat is slow enough that an invented internal cap (e.g. the + // old hard-coded 5s) would be indistinguishable from "no cap" at + // this scale if one were still silently applied; what this proves + // is that a generous caller deadline is honoured in full, not that + // any particular duration is safe. + client.On("Stat", mock.Anything, mock.Anything).Return( + func(ctx context.Context, _ *providerv1beta1.StatRequest, _ ...grpc.CallOption) (*providerv1beta1.StatResponse, error) { + select { + case <-time.After(50 * time.Millisecond): + return &providerv1beta1.StatResponse{ + Status: status.NewOK(ctx), + Info: &providerv1beta1.ResourceInfo{PermissionSet: &providerv1beta1.ResourcePermissions{ListGrants: true}}, + }, nil + case <-ctx.Done(): + return nil, ctx.Err() + } + }) + + generousCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + + shares, err := m.ListPublicShares(generousCtx, user1, nil, false) + Expect(err).ToNot(HaveOccurred()) + // No artificial cap applies: every foreign share is present. + Expect(opaqueIDs(shares)).To(ConsistOf("foreign-0", "foreign-1", "foreign-2", "foreign-3", "foreign-4")) + }) + + It("costs zero Stat calls when the caller created every share", func() { + const numOwn = 10 + specs := make([]shareSpec, 0, numOwn) + for i := 0; i < numOwn; i++ { + specs = append(specs, shareSpec{ + ID: fmt.Sprintf("own-%d", i), + Token: fmt.Sprintf("t-own-%d", i), + Creator: user1.Id, + StorageID: "storageid", + SpaceID: "space-own", + OpaqueID: fmt.Sprintf("o-%d", i), + }) + } + seedShares(tmpFile.Name(), specs) + + shares, err := m.ListPublicShares(ctx, user1, nil, false) + Expect(err).ToNot(HaveOccurred()) + Expect(shares).To(HaveLen(numOwn)) + client.AssertNumberOfCalls(GinkgoT(), "Stat", 0) + }) + + It("still stats when a resource filter is given", func() { + user2 := &userpb.UserId{Idp: "https://localhost:9200", OpaqueId: "einstein"} + rid := &providerv1beta1.ResourceId{StorageId: "storageid", SpaceId: "space-x", OpaqueId: "o1"} + seedShares(tmpFile.Name(), []shareSpec{ + {ID: "foreign-1", Token: "t-f1", Creator: user2, StorageID: "storageid", SpaceID: "space-x", OpaqueID: "o1"}, + }) + client.On("Stat", mock.Anything, mock.Anything).Return(&providerv1beta1.StatResponse{ + Status: status.NewOK(ctx), + Info: &providerv1beta1.ResourceInfo{PermissionSet: &providerv1beta1.ResourcePermissions{ListGrants: true}}, + }, nil) + + shares, err := m.ListPublicShares(ctx, user1, []*link.ListPublicSharesRequest_Filter{publicshare.ResourceIDFilter(rid)}, false) + Expect(err).ToNot(HaveOccurred()) + Expect(shares).To(HaveLen(1)) + client.AssertNumberOfCalls(GinkgoT(), "Stat", 1) + }) + + It("stats a denied resource only once even with several shares on it", func() { + user2 := &userpb.UserId{Idp: "https://localhost:9200", OpaqueId: "einstein"} + rid := &providerv1beta1.ResourceId{StorageId: "storageid", SpaceId: "space-x", OpaqueId: "o1"} + seedShares(tmpFile.Name(), []shareSpec{ + {ID: "f1", Token: "t-f1", Creator: user2, StorageID: "storageid", SpaceID: "space-x", OpaqueID: "o1"}, + {ID: "f2", Token: "t-f2", Creator: user2, StorageID: "storageid", SpaceID: "space-x", OpaqueID: "o1"}, + }) + client.On("Stat", mock.Anything, mock.Anything).Return(&providerv1beta1.StatResponse{ + Status: status.NewOK(ctx), + Info: &providerv1beta1.ResourceInfo{PermissionSet: &providerv1beta1.ResourcePermissions{ListGrants: false}}, + }, nil) + + shares, err := m.ListPublicShares(ctx, user1, []*link.ListPublicSharesRequest_Filter{publicshare.ResourceIDFilter(rid)}, false) + Expect(err).ToNot(HaveOccurred()) + Expect(shares).To(BeEmpty()) + client.AssertNumberOfCalls(GinkgoT(), "Stat", 1) // deduplicated to a single Stat call by the pass-1 resource map + }) + + // userCanListGrants fails closed (excludes the share, records false) for + // every one of these outcomes. None of them had a spec before: a + // regression flipping any one to fail-open (e.g. recording true on an + // error) would not have failed a single test. + DescribeTable("excludes a foreign share when Stat fails closed", + func(statFn func(ctx context.Context) (*providerv1beta1.StatResponse, error)) { + user2 := &userpb.UserId{Idp: "https://localhost:9200", OpaqueId: "einstein"} + seedShares(tmpFile.Name(), []shareSpec{ + {ID: "own-1", Token: "t-own-1", Creator: user1.Id, StorageID: "storageid", SpaceID: "space-own", OpaqueID: "oown"}, + {ID: "foreign-1", Token: "t-f1", Creator: user2, StorageID: "storageid", SpaceID: "space-x", OpaqueID: "o1"}, + }) + + client.On("Stat", mock.Anything, mock.Anything).Return( + func(ctx context.Context, _ *providerv1beta1.StatRequest, _ ...grpc.CallOption) (*providerv1beta1.StatResponse, error) { + return statFn(ctx) + }) + + shares, err := m.ListPublicShares(ctx, user1, nil, false) + Expect(err).ToNot(HaveOccurred()) + // The own share proves this is exclusion of the foreign share, not a + // blanket empty result from some unrelated failure. + Expect(opaqueIDs(shares)).To(ConsistOf("own-1")) + Expect(opaqueIDs(shares)).ToNot(ContainElement("foreign-1")) + }, + Entry("transport error", func(_ context.Context) (*providerv1beta1.StatResponse, error) { + return nil, errors.New("transport error talking to the gateway") + }), + Entry("CODE_NOT_FOUND", func(ctx context.Context) (*providerv1beta1.StatResponse, error) { + return &providerv1beta1.StatResponse{ + Status: status.NewNotFound(ctx, "resource not found"), + }, nil + }), + Entry("other non-OK status", func(ctx context.Context) (*providerv1beta1.StatResponse, error) { + return &providerv1beta1.StatResponse{ + Status: status.NewInternal(ctx, "internal error"), + }, nil + }), + Entry("OK with nil Info", func(ctx context.Context) (*providerv1beta1.StatResponse, error) { + return &providerv1beta1.StatResponse{ + Status: status.NewOK(ctx), + Info: nil, + }, nil + }), + Entry("OK with Info but nil PermissionSet", func(ctx context.Context) (*providerv1beta1.StatResponse, error) { + return &providerv1beta1.StatResponse{ + Status: status.NewOK(ctx), + Info: &providerv1beta1.ResourceInfo{PermissionSet: nil}, + }, nil + }), + ) }) Describe("Load", func() { diff --git a/pkg/publicshare/publicshare.go b/pkg/publicshare/publicshare.go index 61e36f3e049..7174fffccb9 100644 --- a/pkg/publicshare/publicshare.go +++ b/pkg/publicshare/publicshare.go @@ -136,7 +136,7 @@ func MatchesFilter(share *link.PublicShare, filter *link.ListPublicSharesRequest case link.ListPublicSharesRequest_Filter_TYPE_RESOURCE_ID: return utils.ResourceIDEqual(share.ResourceId, filter.GetResourceId()) case StorageIDFilterType: - return share.ResourceId.StorageId == filter.GetResourceId().GetStorageId() + return share.GetResourceId().GetStorageId() == filter.GetResourceId().GetStorageId() default: return false } diff --git a/pkg/publicshare/publicshare_suite_test.go b/pkg/publicshare/publicshare_suite_test.go new file mode 100644 index 00000000000..221c21b9d11 --- /dev/null +++ b/pkg/publicshare/publicshare_suite_test.go @@ -0,0 +1,31 @@ +// Copyright 2018-2021 CERN +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// In applying this license, CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +package publicshare_test + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestPublicshare(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Publicshare Suite") +} diff --git a/pkg/publicshare/publicshare_test.go b/pkg/publicshare/publicshare_test.go new file mode 100644 index 00000000000..a92a162f0f4 --- /dev/null +++ b/pkg/publicshare/publicshare_test.go @@ -0,0 +1,56 @@ +// Copyright 2018-2021 CERN +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// In applying this license, CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +package publicshare_test + +import ( + link "github.com/cs3org/go-cs3apis/cs3/sharing/link/v1beta1" + provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" + "github.com/owncloud/reva/v2/pkg/publicshare" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("MatchesFilters", func() { + // A persisted public share can have a nil ResourceId (see OCISDEV-877: ocis shipped + // a repair CLI for exactly this class of corrupt row). MatchesFilters is called + // before ListPublicShares' own nil-ResourceId guard, so it must not panic. + It("does not panic and returns false for a StorageIDFilter against a share with a nil ResourceId", func() { + share := &link.PublicShare{ResourceId: nil} + filters := []*link.ListPublicSharesRequest_Filter{ + publicshare.StorageIDFilter("s"), + } + + var result bool + Expect(func() { result = publicshare.MatchesFilters(share, filters) }).ToNot(Panic()) + Expect(result).To(BeFalse()) + }) + + It("does not panic and returns false for a TYPE_RESOURCE_ID filter against a share with a nil ResourceId", func() { + share := &link.PublicShare{ResourceId: nil} + rid := &provider.ResourceId{StorageId: "s", SpaceId: "sp", OpaqueId: "o"} + filters := []*link.ListPublicSharesRequest_Filter{ + publicshare.ResourceIDFilter(rid), + } + + var result bool + Expect(func() { result = publicshare.MatchesFilters(share, filters) }).ToNot(Panic()) + Expect(result).To(BeFalse()) + }) +})