From 1dee937e72f948df10e1f49f1fccc3cf43f50ff2 Mon Sep 17 00:00:00 2001 From: micbar Date: Fri, 9 Oct 2026 11:15:17 +0200 Subject: [PATCH] chore: bump reva to latest main --- go.mod | 2 +- go.sum | 4 +- .../v2/pkg/publicshare/manager/json/json.go | 249 ++++++++++++++---- .../reva/v2/pkg/publicshare/publicshare.go | 2 +- .../v2/pkg/share/manager/jsoncs3/jsoncs3.go | 46 ++++ .../manager/jsoncs3/migrations/migration.go | 77 +++++- vendor/modules.txt | 2 +- 7 files changed, 328 insertions(+), 54 deletions(-) diff --git a/go.mod b/go.mod index c5a5b9b644..75decc9ee4 100644 --- a/go.mod +++ b/go.mod @@ -66,7 +66,7 @@ require ( github.com/open-policy-agent/opa v1.21.1 github.com/opencloud-eu/icap-client v0.0.0-20250930132611-28a2afe62d89 github.com/opencloud-eu/libre-graph-api-go v1.0.8 - github.com/opencloud-eu/reva/v2 v2.51.1-0.20261007142032-7148a82e0e45 + github.com/opencloud-eu/reva/v2 v2.51.1-0.20261008105650-ad8e1a1782c0 github.com/opensearch-project/opensearch-go/v4 v4.7.3 github.com/orcaman/concurrent-map v1.0.0 github.com/pkg/errors v0.9.1 diff --git a/go.sum b/go.sum index a0fb3186f1..3426370d38 100644 --- a/go.sum +++ b/go.sum @@ -761,8 +761,8 @@ github.com/opencloud-eu/icap-client v0.0.0-20250930132611-28a2afe62d89 h1:W1ms+l github.com/opencloud-eu/icap-client v0.0.0-20250930132611-28a2afe62d89/go.mod h1:vigJkNss1N2QEceCuNw/ullDehncuJNFB6mEnzfq9UI= github.com/opencloud-eu/libre-graph-api-go v1.0.8 h1:VTkFLkLNna+T4RN6JUZnVQ1bqVAfPtMeORXbFzEYGKs= github.com/opencloud-eu/libre-graph-api-go v1.0.8/go.mod h1:lTM8JeGblNpoMySTW7Lui2+c5TTLI95mwxtdUIHHrhU= -github.com/opencloud-eu/reva/v2 v2.51.1-0.20261007142032-7148a82e0e45 h1:rrZBLsznQDL6YFjtbZF+Z8WCHbGSWn0RsD0ddwaVY8c= -github.com/opencloud-eu/reva/v2 v2.51.1-0.20261007142032-7148a82e0e45/go.mod h1:rAwuI3PX8n3RxBXHfpMtAKqfwxTIKhZol1Bi3Qj39p8= +github.com/opencloud-eu/reva/v2 v2.51.1-0.20261008105650-ad8e1a1782c0 h1:PU1sWQD5PLTziT1EuWt4vuU5Mzm77YtLNCwZOhKXpwM= +github.com/opencloud-eu/reva/v2 v2.51.1-0.20261008105650-ad8e1a1782c0/go.mod h1:rAwuI3PX8n3RxBXHfpMtAKqfwxTIKhZol1Bi3Qj39p8= github.com/opencloud-eu/secure v0.0.0-20260312082735-b6f5cb2244e4 h1:l2oB/RctH+t8r7QBj5p8thfEHCM/jF35aAY3WQ3hADI= github.com/opencloud-eu/secure v0.0.0-20260312082735-b6f5cb2244e4/go.mod h1:BmF5hyM6tXczk3MpQkFf1hpKSRqCyhqcbiQtiAF7+40= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= diff --git a/vendor/github.com/opencloud-eu/reva/v2/pkg/publicshare/manager/json/json.go b/vendor/github.com/opencloud-eu/reva/v2/pkg/publicshare/manager/json/json.go index 721f86a5b7..257ebfbaae 100644 --- a/vendor/github.com/opencloud-eu/reva/v2/pkg/publicshare/manager/json/json.go +++ b/vendor/github.com/opencloud-eu/reva/v2/pkg/publicshare/manager/json/json.go @@ -25,15 +25,16 @@ 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" @@ -50,10 +51,15 @@ import ( "github.com/opencloud-eu/reva/v2/pkg/publicshare/manager/registry" "github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool" "github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata" + "github.com/opencloud-eu/reva/v2/pkg/storagespace" "github.com/opencloud-eu/reva/v2/pkg/utils" "github.com/pkg/errors" + "google.golang.org/protobuf/types/known/fieldmaskpb" ) +// defaultStatConcurrency bounds concurrent Stat RPCs in ListPublicShares, mirroring the decomposedfs share manager. +const defaultStatConcurrency = 5 + func init() { registry.Register("json", NewFile) registry.Register("jsoncs3", NewCS3) @@ -111,6 +117,7 @@ func NewCS3(c map[string]interface{}) (publicshare.Manager, error) { func New(gwAddr string, pwHashCost, janitorRunInterval int, enableCleanup bool, p persistence.Persistence) (publicshare.Manager, error) { m := &manager{ gatewayAddr: gwAddr, + maxConcurrency: defaultStatConcurrency, mutex: &sync.Mutex{}, passwordHashCost: pwHashCost, janitorRunInterval: janitorRunInterval, @@ -154,6 +161,9 @@ func (c *commonConfig) init() { } type manager struct { + // maxConcurrency bounds concurrent Stat RPCs in ListPublicShares. + maxConcurrency int + gatewayAddr string mutex *sync.Mutex persistence persistence.Persistence @@ -481,33 +491,30 @@ func (m *manager) GetPublicShare(ctx context.Context, u *user.User, ref *link.Pu return nil, errtypes.NotFound("no shares found by id:" + ref.GetId().String()) } -// ListPublicShares retrieves all the shares on the manager that are valid. -func (m *manager) ListPublicShares(ctx context.Context, u *user.User, filters []*link.ListPublicSharesRequest_Filter, sign bool) ([]*link.PublicShare, error) { +// collectShares splits the persisted shares into own and foreign ones, plus the distinct resources the foreign ones point at. +func (m *manager) collectShares(ctx context.Context, u *user.User, filters []*link.ListPublicSharesRequest_Filter) ([]*publicShare, []*publicShare, map[string]*provider.ResourceId, error) { m.mutex.Lock() defer m.mutex.Unlock() if err := m.init(); err != nil { - return nil, err + return nil, nil, nil, err } log := appctx.GetLogger(ctx) db, err := m.persistence.Read(ctx) if err != nil { - return nil, err + return nil, nil, 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{}) + 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 { - return nil, err + return nil, nil, nil, err } if publicshare.IsExpired(&local.PublicShare) { @@ -523,51 +530,203 @@ 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 local.ResourceId == nil { + log.Warn(). + Str("share_id", local.PublicShare.GetId().GetOpaqueId()). + Str("share_token", local.Token). + Msg("ListPublicShares: skipping share with nil resource_id") + continue } + if publicshare.IsCreatedByUser(&local.PublicShare, u) { + ownShares = append(ownShares, &local) + continue + } + + foreignShares = append(foreignShares, &local) + foreignResourceIDs[storagespace.FormatResourceID(local.ResourceId)] = local.ResourceId + } + + return ownShares, foreignShares, foreignResourceIDs, nil +} + +// ListPublicShares retrieves all the shares on the manager that are valid. +func (m *manager) ListPublicShares(ctx context.Context, u *user.User, filters []*link.ListPublicSharesRequest_Filter, sign bool) ([]*link.PublicShare, error) { + // collectShares holds the lock; nothing below it touches manager state + ownShares, foreignShares, foreignResourceIDs, err := m.collectShares(ctx, u, filters) + if err != nil { + return nil, err + } + + // stat each distinct foreign resource once, concurrently, within a time budget + 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.PasswordProtected && sign { + if err := publicshare.AddSignature(&local.PublicShare, local.Password); err != nil { + return nil, err + } + } + shares = append(shares, &local.PublicShare) + } + for _, local := range foreignShares { + // fail closed: a resource left undecided is absent here and stays excluded + if !permitted[storagespace.FormatResourceID(local.ResourceId)] { + continue + } if local.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 resource once within a time budget and reports who may list its grants; anything absent was never decided. +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) + + // stop feeding jobs once the budget runs out so workers drain instead of blocking + g.Go(func() error { + defer close(jobs) + for _, rid := range resourceIDs { + select { + case jobs <- job{rid}: + case <-gctx.Done(): + return nil + } + } + return nil + }) + + // workers, bounded by numWorkers <= m.maxConcurrency concurrent Stat RPCs + for i := 0; i < numWorkers; i++ { + g.Go(func() error { + for j := range jobs { + if gctx.Err() != nil { + // budget exhausted: leave the rest undecided, the caller treats them as denied + 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 bounds the stat fan-out to the caller's own deadline minus a margin, or not at all if it has none. +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 keyed by storagespace.FormatResourceID, safe for concurrent use. +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 copies the results collected so far; call it only once no writers remain. +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 stats the resource, records whether the user may list its grants, and reports the answer. +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/vendor/github.com/opencloud-eu/reva/v2/pkg/publicshare/publicshare.go b/vendor/github.com/opencloud-eu/reva/v2/pkg/publicshare/publicshare.go index a62b41e3fa..84f27f8e48 100644 --- a/vendor/github.com/opencloud-eu/reva/v2/pkg/publicshare/publicshare.go +++ b/vendor/github.com/opencloud-eu/reva/v2/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/vendor/github.com/opencloud-eu/reva/v2/pkg/share/manager/jsoncs3/jsoncs3.go b/vendor/github.com/opencloud-eu/reva/v2/pkg/share/manager/jsoncs3/jsoncs3.go index 492ec38009..a10b1d597e 100644 --- a/vendor/github.com/opencloud-eu/reva/v2/pkg/share/manager/jsoncs3/jsoncs3.go +++ b/vendor/github.com/opencloud-eu/reva/v2/pkg/share/manager/jsoncs3/jsoncs3.go @@ -170,6 +170,10 @@ type Manager struct { gatewaySelector pool.Selectable[gatewayv1beta1.GatewayAPIClient] eventStream events.Stream logger *zerolog.Logger + + // migrationCfg is kept so that migrations can be run again later, see + // RunMigrationsFrom. + migrationCfg migration.MigrationConfig } // NewDefault returns a new manager instance with default dependencies @@ -372,9 +376,26 @@ func (m *Manager) migrationsRunning() bool { // called once after New() in production server startup. Callers that do not // need migrations should call SkipMigrations instead to unblock write operations. func (m *Manager) RunMigrations(cfg migration.MigrationConfig) { + m.migrationCfg = cfg go m.doMigrations(cfg) } +// RunMigration runs the named migration whether or not it has already been +// applied, and blocks until it is done. It is meant for operators repeating a +// migration; the ordinary startup path is RunMigrations. +func (m *Manager) RunMigration(ctx context.Context, name string) error { + if err := m.waitForInit(ctx); err != nil { + return err + } + // the startup run holds the migration lock; waiting for it to finish keeps + // this from being silently skipped + if err := m.waitForMigrations(ctx); err != nil { + return err + } + migrations := migration.New(*m.logger, m.gatewaySelector, m.storage, m.migrationCfg, m, m) + return migrations.RunMigration(name) +} + func (m *Manager) waitForMigrations(ctx context.Context) error { select { case <-m.migrationsDone: @@ -660,9 +681,34 @@ func (m *Manager) Unshare(ctx context.Context, ref *collaboration.ShareReference return err } + // only the creator or a user with RemoveGrant on the resource may unshare + user := ctxpkg.ContextMustGetUser(ctx) + if !share.IsCreatedByUser(s, user) && !m.userHasRemoveGrantPermission(ctx, s) { + // TODO why not permission denied? + return errtypes.NotFound(ref.String()) + } + return m.removeShare(ctx, s, false) } +// userHasRemoveGrantPermission reports whether the context user holds RemoveGrant on the share's resource. +func (m *Manager) userHasRemoveGrantPermission(ctx context.Context, s *collaboration.Share) bool { + req := &provider.StatRequest{ + Ref: &provider.Reference{ResourceId: s.ResourceId}, + FieldMask: &fieldmaskpb.FieldMask{ + Paths: []string{"permissions"}, + }, + } + client, err := m.gatewaySelector.Next() + if err != nil { + return false + } + res, err := client.Stat(ctx, req) + return err == nil && + res.Status.Code == rpcv1beta1.Code_CODE_OK && + res.Info.PermissionSet.RemoveGrant +} + // UpdateShare updates the mode of the given share. func (m *Manager) UpdateShare(ctx context.Context, ref *collaboration.ShareReference, p *collaboration.SharePermissions, updated *collaboration.Share, fieldMask *field_mask.FieldMask) (*collaboration.Share, error) { ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "UpdateShare") diff --git a/vendor/github.com/opencloud-eu/reva/v2/pkg/share/manager/jsoncs3/migrations/migration.go b/vendor/github.com/opencloud-eu/reva/v2/pkg/share/manager/jsoncs3/migrations/migration.go index e05bfb9676..3531fbaac3 100644 --- a/vendor/github.com/opencloud-eu/reva/v2/pkg/share/manager/jsoncs3/migrations/migration.go +++ b/vendor/github.com/opencloud-eu/reva/v2/pkg/share/manager/jsoncs3/migrations/migration.go @@ -27,6 +27,7 @@ import ( "os" "os/signal" "slices" + "strings" "syscall" "time" @@ -347,14 +348,80 @@ func (m *Migrations) saveState(ctx context.Context) error { return m.storage.SimpleUpload(ctx, stateFile, data) } +// RunMigrations runs every migration that has not been applied yet, as +// recorded in the share manager's state. func (m *Migrations) RunMigrations() { + // errors are logged by runMigrations; the startup path has nowhere to + // report them to and must not block the manager from coming up + _ = m.runMigrations() +} + +// Names returns the names of the registered migrations, in the order they +// would be applied. +func Names() []string { + names := make([]string, 0, len(migrations)) + for _, mig := range migrations { + names = append(names, mig.Name()) + } + return names +} + +// RunMigration runs the named migration, whether or not it has been applied +// before. It exists so that an operator can repeat a migration: the +// import_space_members migration for instance imports space grants that were +// written outside of the share manager, and has to be repeatable. +// +// The recorded version is raised if the migration is newer than it, and never +// lowered. +func (m *Migrations) RunMigration(name string) error { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + var mig migration + for _, candidate := range migrations { + if candidate.Name() == name { + mig = candidate + break + } + } + if mig == nil { + return fmt.Errorf("unknown migration %q, available: %s", name, strings.Join(Names(), ", ")) + } + + etag, err := m.acquireLock(ctx) + if err != nil { + return err + } + cancelHB := m.startHeartbeat(ctx, etag) + defer cancelHB() + defer m.releaseLock(ctx) + + if err := m.loadState(ctx); err != nil { + return err + } + + m.logger.Info().Str("migration", mig.Name()).Int("version", mig.Version()).Msg("running migration") + mig.Initialize(m.config) + if err := mig.Migrate(); err != nil { + return err + } + + if mig.Version() > m.state.version { + m.state.version = mig.Version() + return m.saveState(ctx) + } + return nil +} + +// runMigrations applies the migrations above the current base version.// runMigrations applies the migrations above the current base version. +func (m *Migrations) runMigrations() error { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() etag, err := m.acquireLock(ctx) if err != nil { m.logger.Error().Err(err).Msg("failed to acquire migration lock; skipping migrations") - return + return err } cancelHB := m.startHeartbeat(ctx, etag) defer cancelHB() @@ -362,7 +429,7 @@ func (m *Migrations) RunMigrations() { if err := m.loadState(ctx); err != nil { m.logger.Error().Err(err).Msg("failed to load migration state; skipping migrations") - return + return err } m.logger.Info().Int("current state", m.state.version).Msg("checking migrations") @@ -373,15 +440,17 @@ func (m *Migrations) RunMigrations() { mig.Initialize(m.config) if err := mig.Migrate(); err != nil { m.logger.Error().Err(err).Str("migration", mig.Name()).Msg("migration failed; stopping") - return + return err } m.state.version = mig.Version() if err := m.saveState(ctx); err != nil { m.logger.Error().Err(err).Msg("failed to save migration state; stopping") - return + return err } } else { m.logger.Info().Str("migration", mig.Name()).Int("version", mig.Version()).Msg("skipping migration") } } + + return nil } diff --git a/vendor/modules.txt b/vendor/modules.txt index a538cb6a79..436b3d5e23 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1360,7 +1360,7 @@ github.com/opencloud-eu/icap-client # github.com/opencloud-eu/libre-graph-api-go v1.0.8 ## explicit; go 1.23 github.com/opencloud-eu/libre-graph-api-go -# github.com/opencloud-eu/reva/v2 v2.51.1-0.20261007142032-7148a82e0e45 +# github.com/opencloud-eu/reva/v2 v2.51.1-0.20261008105650-ad8e1a1782c0 ## explicit; go 1.26.0 github.com/opencloud-eu/reva/v2/cmd/revad/internal/grace github.com/opencloud-eu/reva/v2/cmd/revad/runtime