Merge pull request #3685 from opencloud-eu/bump-reva

chore: bump reva to latest main

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

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=

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()

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
 	}

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")

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
 }

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