From 5362f82efaf78ae25496626f91171315bf7676f2 Mon Sep 17 00:00:00 2001 From: kobergj Date: Fri, 22 Dec 2023 15:10:39 +0100 Subject: [PATCH] bump reva (#8058) Signed-off-by: jkoberg --- changelog/unreleased/bump-reva.md | 1 + go.mod | 2 +- go.sum | 4 +- .../publicshareprovider.go | 7 +- .../sharesstorageprovider.go | 22 +- .../usershareprovider/usershareprovider.go | 66 +- .../v2/pkg/rhttp/datatx/manager/tus/tus.go | 24 +- .../receivedsharecache/receivedsharecache.go | 43 +- .../utils/decomposedfs/decomposedfs.go | 66 +- .../storage/utils/decomposedfs/node/node.go | 25 +- .../storage/utils/decomposedfs/revisions.go | 2 +- .../pkg/storage/utils/decomposedfs/spaces.go | 2 +- .../pkg/storage/utils/decomposedfs/upload.go | 260 ++++---- .../utils/decomposedfs/upload/processing.go | 582 ------------------ .../utils/decomposedfs/upload/session.go | 319 ++++++++++ .../utils/decomposedfs/upload/store.go | 380 ++++++++++++ .../utils/decomposedfs/upload/upload.go | 270 +++----- vendor/modules.txt | 2 +- 18 files changed, 1130 insertions(+), 947 deletions(-) delete mode 100644 vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/processing.go create mode 100644 vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/session.go create mode 100644 vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/store.go diff --git a/changelog/unreleased/bump-reva.md b/changelog/unreleased/bump-reva.md index f7e1fadc8..4d6281697 100644 --- a/changelog/unreleased/bump-reva.md +++ b/changelog/unreleased/bump-reva.md @@ -3,3 +3,4 @@ Enhancement: Bump reva Bumps reva version https://github.com/owncloud/ocis/pull/8038 +https://github.com/owncloud/ocis/pull/8056 diff --git a/go.mod b/go.mod index 1ecc0ab34..43d2636cd 100644 --- a/go.mod +++ b/go.mod @@ -13,7 +13,7 @@ require ( github.com/coreos/go-oidc v2.2.1+incompatible github.com/coreos/go-oidc/v3 v3.9.0 github.com/cs3org/go-cs3apis v0.0.0-20231023073225-7748710e0781 - github.com/cs3org/reva/v2 v2.17.1-0.20231220115644-93b4dd91a8b3 + github.com/cs3org/reva/v2 v2.17.1-0.20231222094355-457e00743c93 github.com/dhowden/tag v0.0.0-20230630033851-978a0926ee25 github.com/disintegration/imaging v1.6.2 github.com/dutchcoders/go-clamd v0.0.0-20170520113014-b970184f4d9e diff --git a/go.sum b/go.sum index c4b1442f0..8aa7ae312 100644 --- a/go.sum +++ b/go.sum @@ -1021,8 +1021,8 @@ github.com/crewjam/saml v0.4.14 h1:g9FBNx62osKusnFzs3QTN5L9CVA/Egfgm+stJShzw/c= github.com/crewjam/saml v0.4.14/go.mod h1:UVSZCf18jJkk6GpWNVqcyQJMD5HsRugBPf4I1nl2mME= github.com/cs3org/go-cs3apis v0.0.0-20231023073225-7748710e0781 h1:BUdwkIlf8IS2FasrrPg8gGPHQPOrQ18MS1Oew2tmGtY= github.com/cs3org/go-cs3apis v0.0.0-20231023073225-7748710e0781/go.mod h1:UXha4TguuB52H14EMoSsCqDj7k8a/t7g4gVP+bgY5LY= -github.com/cs3org/reva/v2 v2.17.1-0.20231220115644-93b4dd91a8b3 h1:9vKLWy7yK/I4MinF5AARDsM4lgaALgZoonEdAf972Pc= -github.com/cs3org/reva/v2 v2.17.1-0.20231220115644-93b4dd91a8b3/go.mod h1:QW31Q1IQ9ZCJMFv3u8/SdHSyLfCcSVNcRbqIJj+Y+7o= +github.com/cs3org/reva/v2 v2.17.1-0.20231222094355-457e00743c93 h1:kOf1+hIKFGeRBTiDgHOgjEHT3FwDfPkc+QJMJ9oE1Ds= +github.com/cs3org/reva/v2 v2.17.1-0.20231222094355-457e00743c93/go.mod h1:QW31Q1IQ9ZCJMFv3u8/SdHSyLfCcSVNcRbqIJj+Y+7o= github.com/cyberdelia/templates v0.0.0-20141128023046-ca7fffd4298c/go.mod h1:GyV+0YP4qX0UQ7r2MoYZ+AvYDp12OF5yg4q8rGnyNh4= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= diff --git a/vendor/github.com/cs3org/reva/v2/internal/grpc/services/publicshareprovider/publicshareprovider.go b/vendor/github.com/cs3org/reva/v2/internal/grpc/services/publicshareprovider/publicshareprovider.go index 4cfd98cb2..201b2c0f9 100644 --- a/vendor/github.com/cs3org/reva/v2/internal/grpc/services/publicshareprovider/publicshareprovider.go +++ b/vendor/github.com/cs3org/reva/v2/internal/grpc/services/publicshareprovider/publicshareprovider.go @@ -241,8 +241,10 @@ func (s *service) CreatePublicShare(ctx context.Context, req *link.CreatePublicS }, nil } - // check if the user can share with the desired permissions - if !conversions.SufficientCS3Permissions(sRes.GetInfo().GetPermissionSet(), req.GetGrant().GetPermissions().GetPermissions()) { + // check if the user can share with the desired permissions. For internal links this is skipped, + // users can always create internal links provided they have the AddGrant permission, which was already + // checked above + if !isInternalLink && !conversions.SufficientCS3Permissions(sRes.GetInfo().GetPermissionSet(), req.GetGrant().GetPermissions().GetPermissions()) { return &link.CreatePublicShareResponse{ Status: status.NewInvalidArg(ctx, "insufficient permissions to create that kind of share"), }, nil @@ -512,6 +514,7 @@ func (s *service) UpdatePublicShare(ctx context.Context, req *link.UpdatePublicS // check if the user can change the permissions to the desired permissions updatePermissions := req.GetUpdate().GetType() == link.UpdatePublicShareRequest_Update_TYPE_PERMISSIONS if updatePermissions && + !isInternalLink && !conversions.SufficientCS3Permissions( sRes.GetInfo().GetPermissionSet(), req.GetUpdate().GetGrant().GetPermissions().GetPermissions(), diff --git a/vendor/github.com/cs3org/reva/v2/internal/grpc/services/sharesstorageprovider/sharesstorageprovider.go b/vendor/github.com/cs3org/reva/v2/internal/grpc/services/sharesstorageprovider/sharesstorageprovider.go index 207992f51..40b48be42 100644 --- a/vendor/github.com/cs3org/reva/v2/internal/grpc/services/sharesstorageprovider/sharesstorageprovider.go +++ b/vendor/github.com/cs3org/reva/v2/internal/grpc/services/sharesstorageprovider/sharesstorageprovider.go @@ -302,7 +302,27 @@ func (s *service) GetPath(ctx context.Context, req *provider.GetPathRequest) (*p }, nil } - return nil, gstatus.Errorf(codes.Unimplemented, "method not implemented") + receivedShare, rpcStatus, err := s.resolveAcceptedShare(ctx, &provider.Reference{ + ResourceId: req.ResourceId, + }) + appctx.GetLogger(ctx).Debug(). + Interface("resourceId", req.ResourceId). + Interface("received_share", receivedShare). + Msg("sharesstorageprovider: Got GetPath request") + if err != nil { + return nil, err + } + if rpcStatus.Code != rpc.Code_CODE_OK { + return &provider.GetPathResponse{ + Status: rpcStatus, + }, nil + } + + return &provider.GetPathResponse{ + Status: status.NewOK(ctx), + Path: receivedShare.MountPoint.Path, + }, nil + } func (s *service) GetHome(ctx context.Context, req *provider.GetHomeRequest) (*provider.GetHomeResponse, error) { diff --git a/vendor/github.com/cs3org/reva/v2/internal/grpc/services/usershareprovider/usershareprovider.go b/vendor/github.com/cs3org/reva/v2/internal/grpc/services/usershareprovider/usershareprovider.go index 5e09a084e..6efc53a59 100644 --- a/vendor/github.com/cs3org/reva/v2/internal/grpc/services/usershareprovider/usershareprovider.go +++ b/vendor/github.com/cs3org/reva/v2/internal/grpc/services/usershareprovider/usershareprovider.go @@ -154,6 +154,7 @@ func (s *service) isPathAllowed(path string) bool { } func (s *service) CreateShare(ctx context.Context, req *collaboration.CreateShareRequest) (*collaboration.CreateShareResponse, error) { + log := appctx.GetLogger(ctx) user := ctxpkg.ContextMustGetUser(ctx) gatewayClient, err := s.gatewaySelector.Next() @@ -184,9 +185,22 @@ func (s *service) CreateShare(ctx context.Context, req *collaboration.CreateShar } } + sRes, err := gatewayClient.Stat(ctx, &provider.StatRequest{Ref: &provider.Reference{ResourceId: req.GetResourceInfo().GetId()}}) + if err != nil { + log.Err(err).Interface("resource_id", req.GetResourceInfo().GetId()).Msg("failed to stat resource to share") + return &collaboration.CreateShareResponse{ + Status: status.NewInternal(ctx, "failed to stat shared resource"), + }, err + } + // the user needs to have the AddGrant permissions on the Resource to be able to create a share + if !sRes.GetInfo().GetPermissionSet().AddGrant { + return &collaboration.CreateShareResponse{ + Status: status.NewPermissionDenied(ctx, nil, "no permission to add grants on shared resource"), + }, err + } // check if the requested share creation has sufficient permissions to do so. if shareCreationAllowed := conversions.SufficientCS3Permissions( - req.GetResourceInfo().GetPermissionSet(), + sRes.GetInfo().GetPermissionSet(), req.GetGrant().GetPermissions().GetPermissions(), ); !shareCreationAllowed { return &collaboration.CreateShareResponse{ @@ -214,6 +228,8 @@ func (s *service) CreateShare(ctx context.Context, req *collaboration.CreateShar } func (s *service) RemoveShare(ctx context.Context, req *collaboration.RemoveShareRequest) (*collaboration.RemoveShareResponse, error) { + log := appctx.GetLogger(ctx) + user := ctxpkg.ContextMustGetUser(ctx) share, err := s.sm.GetShare(ctx, req.Ref) if err != nil { return &collaboration.RemoveShareResponse{ @@ -221,6 +237,29 @@ func (s *service) RemoveShare(ctx context.Context, req *collaboration.RemoveShar }, nil } + gatewayClient, err := s.gatewaySelector.Next() + if err != nil { + return nil, err + } + sRes, err := gatewayClient.Stat(ctx, &provider.StatRequest{Ref: &provider.Reference{ResourceId: share.GetResourceId()}}) + if err != nil { + log.Err(err).Interface("resource_id", share.GetResourceId()).Msg("failed to stat shared resource") + return &collaboration.RemoveShareResponse{ + Status: status.NewInternal(ctx, "failed to stat shared resource"), + }, err + } + // the requesting user needs to be either the Owner/Creator of the share or have the RemoveGrant permissions on the Resource + switch { + case utils.UserEqual(user.GetId(), share.GetCreator()) || utils.UserEqual(user.GetId(), share.GetOwner()): + fallthrough + case sRes.GetInfo().GetPermissionSet().RemoveGrant: + break + default: + return &collaboration.RemoveShareResponse{ + Status: status.NewPermissionDenied(ctx, nil, "no permission to remove grants on shared resource"), + }, err + } + err = s.sm.Unshare(ctx, req.Ref) if err != nil { return &collaboration.RemoveShareResponse{ @@ -279,6 +318,7 @@ func (s *service) ListShares(ctx context.Context, req *collaboration.ListSharesR func (s *service) UpdateShare(ctx context.Context, req *collaboration.UpdateShareRequest) (*collaboration.UpdateShareResponse, error) { log := appctx.GetLogger(ctx) + user := ctxpkg.ContextMustGetUser(ctx) gatewayClient, err := s.gatewaySelector.Next() if err != nil { return nil, err @@ -326,6 +366,17 @@ func (s *service) UpdateShare(ctx context.Context, req *collaboration.UpdateShar Status: status.NewInternal(ctx, "failed to stat shared resource"), }, err } + // the requesting user needs to be either the Owner/Creator of the share or have the UpdateGrant permissions on the Resource + switch { + case utils.UserEqual(user.GetId(), currentShare.GetCreator()) || utils.UserEqual(user.GetId(), currentShare.GetOwner()): + fallthrough + case sRes.GetInfo().GetPermissionSet().UpdateGrant: + break + default: + return &collaboration.UpdateShareResponse{ + Status: status.NewPermissionDenied(ctx, nil, "no permission to remove grants on shared resource"), + }, err + } // If this is a permissions update, check if user's permissions on the resource are sufficient to set the desired permissions var newPermissions *provider.ResourcePermissions @@ -399,9 +450,16 @@ func (s *service) GetReceivedShare(ctx context.Context, req *collaboration.GetRe share, err := s.sm.GetReceivedShare(ctx, req.Ref) if err != nil { log.Err(err).Msg("error getting received share") - return &collaboration.GetReceivedShareResponse{ - Status: status.NewInternal(ctx, "error getting received share"), - }, nil + switch err.(type) { + case errtypes.NotFound: + return &collaboration.GetReceivedShareResponse{ + Status: status.NewNotFound(ctx, "error getting received share"), + }, nil + default: + return &collaboration.GetReceivedShareResponse{ + Status: status.NewInternal(ctx, "error getting received share"), + }, nil + } } res := &collaboration.GetReceivedShareResponse{ diff --git a/vendor/github.com/cs3org/reva/v2/pkg/rhttp/datatx/manager/tus/tus.go b/vendor/github.com/cs3org/reva/v2/pkg/rhttp/datatx/manager/tus/tus.go index fbad5e5bb..6da174015 100644 --- a/vendor/github.com/cs3org/reva/v2/pkg/rhttp/datatx/manager/tus/tus.go +++ b/vendor/github.com/cs3org/reva/v2/pkg/rhttp/datatx/manager/tus/tus.go @@ -37,7 +37,6 @@ import ( "github.com/cs3org/reva/v2/pkg/rhttp/datatx/metrics" "github.com/cs3org/reva/v2/pkg/storage" "github.com/cs3org/reva/v2/pkg/storage/cache" - "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload" "github.com/cs3org/reva/v2/pkg/storagespace" "github.com/mitchellh/mapstructure" ) @@ -100,22 +99,25 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) { return nil, err } - if _, ok := fs.(storage.UploadSessionLister); ok { + if usl, ok := fs.(storage.UploadSessionLister); ok { // We can currently only send updates if the fs is decomposedfs as we read very specific keys from the storage map of the tus info go func() { for { ev := <-handler.CompleteUploads // We should be able to get the upload progress with fs.GetUploadProgress, but currently tus will erase the info files // so we create a Progress instance here that is used to read the correct properties - up := upload.Progress{ - Info: ev.Upload, - } - executant := up.Executant() - ref := up.Reference() - datatx.InvalidateCache(&executant, &ref, m.statCache) - if m.publisher != nil { - if err := datatx.EmitFileUploadedEvent(up.SpaceOwner(), &executant, &ref, m.publisher); err != nil { - appctx.GetLogger(context.Background()).Error().Err(err).Msg("failed to publish FileUploaded event") + ups, err := usl.ListUploadSessions(context.Background(), storage.UploadSessionFilter{ID: &ev.Upload.ID}) + if err != nil { + appctx.GetLogger(context.Background()).Error().Err(err).Str("session", ev.Upload.ID).Msg("failed to list upload session") + } else { + up := ups[0] + executant := up.Executant() + ref := up.Reference() + datatx.InvalidateCache(&executant, &ref, m.statCache) + if m.publisher != nil { + if err := datatx.EmitFileUploadedEvent(up.SpaceOwner(), &executant, &ref, m.publisher); err != nil { + appctx.GetLogger(context.Background()).Error().Err(err).Msg("failed to publish FileUploaded event") + } } } } diff --git a/vendor/github.com/cs3org/reva/v2/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go b/vendor/github.com/cs3org/reva/v2/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go index 65111e7e7..1080a808d 100644 --- a/vendor/github.com/cs3org/reva/v2/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go +++ b/vendor/github.com/cs3org/reva/v2/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go @@ -32,6 +32,7 @@ import ( provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" "github.com/cs3org/reva/v2/pkg/appctx" "github.com/cs3org/reva/v2/pkg/errtypes" + "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/mtimesyncedcache" "github.com/cs3org/reva/v2/pkg/storage/utils/metadata" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" @@ -46,7 +47,7 @@ const tracerName = "receivedsharecache" type Cache struct { lockMap sync.Map - ReceivedSpaces map[string]*Spaces + ReceivedSpaces mtimesyncedcache.Map[string, *Spaces] storage metadata.Storage ttl time.Duration @@ -74,7 +75,7 @@ type State struct { // New returns a new Cache instance func New(s metadata.Storage, ttl time.Duration) Cache { return Cache{ - ReceivedSpaces: map[string]*Spaces{}, + ReceivedSpaces: mtimesyncedcache.Map[string, *Spaces]{}, storage: s, ttl: ttl, lockMap: sync.Map{}, @@ -97,7 +98,7 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati span.SetAttributes(attribute.String("cs3.userid", userID)) defer unlock() - if c.ReceivedSpaces[userID] == nil { + if _, ok := c.ReceivedSpaces.Load(userID); !ok { err := c.syncWithLock(ctx, userID) if err != nil { return err @@ -111,7 +112,8 @@ func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaborati persistFunc := func() error { c.initializeIfNeeded(userID, spaceID) - receivedSpace := c.ReceivedSpaces[userID].Spaces[spaceID] + rss, _ := c.ReceivedSpaces.Load(userID) + receivedSpace := rss.Spaces[spaceID] if receivedSpace.States == nil { receivedSpace.States = map[string]*State{} } @@ -171,10 +173,11 @@ func (c *Cache) Get(ctx context.Context, userID, spaceID, shareID string) (*Stat if err != nil { return nil, err } - if c.ReceivedSpaces[userID] == nil || c.ReceivedSpaces[userID].Spaces[spaceID] == nil { + rss, ok := c.ReceivedSpaces.Load(userID) + if !ok || rss.Spaces[spaceID] == nil { return nil, nil } - return c.ReceivedSpaces[userID].Spaces[spaceID].States[shareID], nil + return rss.Spaces[spaceID].States[shareID], nil } // List returns a list of received shares for a given user @@ -192,7 +195,8 @@ func (c *Cache) List(ctx context.Context, userID string) (map[string]*Space, err } spaces := map[string]*Space{} - for spaceID, space := range c.ReceivedSpaces[userID].Spaces { + rss, _ := c.ReceivedSpaces.Load(userID) + for spaceID, space := range rss.Spaces { spaceCopy := &Space{ States: map[string]*State{}, } @@ -220,9 +224,10 @@ func (c *Cache) syncWithLock(ctx context.Context, userID string) error { jsonPath := userJSONPath(userID) span.AddEvent("updating cache") // - update cached list of created shares for the user in memory if changed + rss, _ := c.ReceivedSpaces.Load(userID) dlres, err := c.storage.Download(ctx, metadata.DownloadRequest{ Path: jsonPath, - IfNoneMatch: []string{c.ReceivedSpaces[userID].etag}, + IfNoneMatch: []string{rss.etag}, }) switch err.(type) { case nil: @@ -248,7 +253,7 @@ func (c *Cache) syncWithLock(ctx context.Context, userID string) error { } newSpaces.etag = dlres.Etag - c.ReceivedSpaces[userID] = newSpaces + c.ReceivedSpaces.Store(userID, newSpaces) span.SetStatus(codes.Ok, "") return nil } @@ -259,12 +264,13 @@ func (c *Cache) persist(ctx context.Context, userID string) error { defer span.End() span.SetAttributes(attribute.String("cs3.userid", userID)) - if c.ReceivedSpaces[userID] == nil { + rss, ok := c.ReceivedSpaces.Load(userID) + if !ok { span.SetStatus(codes.Ok, "no received shares") return nil } - createdBytes, err := json.Marshal(c.ReceivedSpaces[userID]) + createdBytes, err := json.Marshal(rss) if err != nil { span.RecordError(err) span.SetStatus(codes.Error, err.Error()) @@ -280,11 +286,11 @@ func (c *Cache) persist(ctx context.Context, userID string) error { ur := metadata.UploadRequest{ Path: jsonPath, Content: createdBytes, - IfMatchEtag: c.ReceivedSpaces[userID].etag, + IfMatchEtag: rss.etag, } // when there is no etag in memory make sure the file has not been created on the server, see https://www.rfc-editor.org/rfc/rfc9110#field.if-match // > If the field value is "*", the condition is false if the origin server has a current representation for the target resource. - if c.ReceivedSpaces[userID].etag == "" { + if rss.etag == "" { ur.IfNoneMatch = []string{"*"} } @@ -303,12 +309,9 @@ func userJSONPath(userID string) string { } func (c *Cache) initializeIfNeeded(userID, spaceID string) { - if c.ReceivedSpaces[userID] == nil { - c.ReceivedSpaces[userID] = &Spaces{ - Spaces: map[string]*Space{}, - } - } - if spaceID != "" && c.ReceivedSpaces[userID].Spaces[spaceID] == nil { - c.ReceivedSpaces[userID].Spaces[spaceID] = &Space{} + rss, _ := c.ReceivedSpaces.LoadOrStore(userID, &Spaces{Spaces: map[string]*Space{}}) + if spaceID != "" && rss.Spaces[spaceID] == nil { + rss.Spaces[spaceID] = &Space{} + c.ReceivedSpaces.Store(userID, rss) } } diff --git a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/decomposedfs.go b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/decomposedfs.go index b6ee40d29..bca49ef18 100644 --- a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/decomposedfs.go +++ b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/decomposedfs.go @@ -62,6 +62,7 @@ import ( "github.com/cs3org/reva/v2/pkg/utils" "github.com/jellydator/ttlcache/v2" "github.com/pkg/errors" + tusd "github.com/tus/tusd/pkg/handler" microstore "go-micro.dev/v4/store" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/trace" @@ -104,6 +105,26 @@ type Tree interface { Propagate(ctx context.Context, node *node.Node, sizeDiff int64) (err error) } +// Session is the interface that OcisSession implements. By combining tus.Upload, +// storage.UploadSession and custom functions we can reuse the same struct throughout +// the whole upload lifecycle. +// +// Some functions that are only used by decomposedfs are not yet part of this interface. +// They might be added after more refactoring. +type Session interface { + tusd.Upload + storage.UploadSession + upload.Session + LockID() string +} + +type SessionStore interface { + New(ctx context.Context) *upload.OcisSession + List(ctx context.Context) ([]*upload.OcisSession, error) + Get(ctx context.Context, id string) (*upload.OcisSession, error) + Cleanup(ctx context.Context, session upload.Session, failure bool, keepUpload bool) +} + // Decomposedfs provides the base for decomposed filesystem implementations type Decomposedfs struct { lu *lookup.Lookup @@ -113,6 +134,7 @@ type Decomposedfs struct { chunkHandler *chunking.ChunkHandler stream events.Stream cache cache.StatCache + sessionStore SessionStore UserCache *ttlcache.Cache userSpaceIndex *spaceidindex.Index @@ -211,6 +233,7 @@ func New(o *options.Options, lu *lookup.Lookup, p Permissions, tp Tree, es event userSpaceIndex: userSpaceIndex, groupSpaceIndex: groupSpaceIndex, spaceTypeIndex: spaceTypeIndex, + sessionStore: upload.NewSessionStore(lu, tp, o.Root, es, o.AsyncFileUploads, o.Tokens), } if o.AsyncFileUploads { @@ -245,7 +268,7 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) { for event := range ch { switch ev := event.Event.(type) { case events.PostprocessingFinished: - up, err := upload.Get(ctx, ev.UploadID, fs.lu, fs.tp, fs.o.Root, fs.stream, fs.o.AsyncFileUploads, fs.o.Tokens) + session, err := fs.sessionStore.Get(ctx, ev.UploadID) if err != nil { log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("Failed to get upload") continue // NOTE: since we can't get the upload, we can't delete the blob @@ -256,12 +279,11 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) { keepUpload bool ) - n, err := node.ReadNode(ctx, fs.lu, up.Info.Storage["SpaceRoot"], up.Info.Storage["NodeId"], false, nil, true) + n, err := session.Node(ctx) if err != nil { log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("could not read node") continue } - up.Node = n switch ev.Outcome { default: @@ -272,7 +294,7 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) { keepUpload = true metrics.UploadSessionsAborted.Inc() case events.PPOutcomeContinue: - if err := up.Finalize(); err != nil { + if err := session.Finalize(); err != nil { log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("could not finalize upload") keepUpload = true // should we keep the upload when assembling failed? failed = true @@ -285,7 +307,7 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) { } getParent := func() *node.Node { - p, err := up.Node.Parent(ctx) + p, err := n.Parent(ctx) if err != nil { log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("could not read parent") return nil @@ -296,7 +318,7 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) { now := time.Now() if failed { // propagate sizeDiff after failed postprocessing - if err := fs.tp.Propagate(ctx, up.Node, -up.SizeDiff); err != nil { + if err := fs.tp.Propagate(ctx, n, -session.SizeDiff()); err != nil { log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("could not propagate tree size change") } } else if p := getParent(); p != nil { @@ -307,7 +329,7 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) { } } - upload.Cleanup(up, failed, keepUpload) + fs.sessionStore.Cleanup(ctx, session, failed, keepUpload) // remove cache entry in gateway fs.cache.RemoveStatContext(ctx, ev.ExecutingUser.GetId(), &provider.ResourceId{SpaceId: n.SpaceID, OpaqueId: n.ID}) @@ -322,11 +344,11 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) { Filename: ev.Filename, FileRef: &provider.Reference{ ResourceId: &provider.ResourceId{ - StorageId: up.Info.MetaData["providerID"], - SpaceId: up.Info.Storage["SpaceRoot"], - OpaqueId: up.Info.Storage["SpaceRoot"], + StorageId: session.ProviderID(), + SpaceId: session.SpaceID(), + OpaqueId: session.SpaceID(), }, - Path: utils.MakeRelativePath(filepath.Join(up.Info.MetaData["dir"], up.Info.MetaData["filename"])), + Path: utils.MakeRelativePath(filepath.Join(session.Dir(), session.Filename())), }, Timestamp: utils.TimeToTS(now), SpaceOwner: n.SpaceOwnerOrManager(ctx), @@ -335,17 +357,17 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) { log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("Failed to publish UploadReady event") } case events.RestartPostprocessing: - up, err := upload.Get(ctx, ev.UploadID, fs.lu, fs.tp, fs.o.Root, fs.stream, fs.o.AsyncFileUploads, fs.o.Tokens) + session, err := fs.sessionStore.Get(ctx, ev.UploadID) if err != nil { log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("Failed to get upload") continue } - n, err := node.ReadNode(ctx, fs.lu, up.Info.Storage["SpaceRoot"], up.Info.Storage["NodeId"], false, nil, true) + n, err := session.Node(ctx) if err != nil { log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("could not read node") continue } - s, err := up.URL(up.Ctx) + s, err := session.URL(ctx) if err != nil { log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("could not create url") continue @@ -355,13 +377,13 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) { // restart postprocessing if err := events.Publish(ctx, fs.stream, events.BytesReceived{ - UploadID: up.Info.ID, + UploadID: session.ID(), URL: s, - SpaceOwner: n.SpaceOwnerOrManager(up.Ctx), + SpaceOwner: n.SpaceOwnerOrManager(ctx), ExecutingUser: &user.User{Id: &user.UserId{OpaqueId: "postprocessing-restart"}}, // send nil instead? ResourceID: &provider.ResourceId{SpaceId: n.SpaceID, OpaqueId: n.ID}, - Filename: up.Info.Storage["NodeName"], - Filesize: uint64(up.Info.Size), + Filename: session.Filename(), + Filesize: uint64(session.Size()), }); err != nil { log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("Failed to publish BytesReceived event") } @@ -460,19 +482,17 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) { */ default: // uploadid is not empty -> this is an async upload - up, err := upload.Get(ctx, ev.UploadID, fs.lu, fs.tp, fs.o.Root, fs.stream, fs.o.AsyncFileUploads, fs.o.Tokens) + session, err := fs.sessionStore.Get(ctx, ev.UploadID) if err != nil { log.Error().Err(err).Str("uploadID", ev.UploadID).Msg("Failed to get upload") continue } - no, err := node.ReadNode(up.Ctx, fs.lu, up.Info.Storage["SpaceRoot"], up.Info.Storage["NodeId"], false, nil, false) + n, err = session.Node(ctx) if err != nil { log.Error().Err(err).Interface("uploadID", ev.UploadID).Msg("Failed to get node after scan") continue } - - n = no } if err := n.SetScanData(ctx, res.Description, res.Scandate); err != nil { @@ -1031,7 +1051,7 @@ func (fs *Decomposedfs) Download(ctx context.Context, ref *provider.Reference) ( if err != nil { return nil, errors.Wrap(err, "Decomposedfs: error getting mtime for '"+n.ID+"'") } - currentEtag, err := node.CalculateEtag(n, mtime) + currentEtag, err := node.CalculateEtag(n.ID, mtime) if err != nil { return nil, errors.Wrap(err, "Decomposedfs: error calculating etag for '"+n.ID+"'") } diff --git a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node/node.go b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node/node.go index a5f9c2b1d..3a9f259d9 100644 --- a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node/node.go +++ b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node/node.go @@ -512,14 +512,9 @@ func (n *Node) LockFilePath() string { } // CalculateEtag returns a hash of fileid + tmtime (or mtime) -func CalculateEtag(n *Node, tmTime time.Time) (string, error) { - return calculateEtag(n, tmTime) -} - -// calculateEtag returns a hash of fileid + tmtime (or mtime) -func calculateEtag(n *Node, tmTime time.Time) (string, error) { +func CalculateEtag(id string, tmTime time.Time) (string, error) { h := md5.New() - if _, err := io.WriteString(h, n.ID); err != nil { + if _, err := io.WriteString(h, id); err != nil { return "", err } /* TODO we could strengthen the etag by adding the blobid, but then all etags would change. we would need a legacy etag check as well @@ -562,7 +557,7 @@ func (n *Node) SetEtag(ctx context.Context, val string) (err error) { return } var etag string - if etag, err = calculateEtag(n, tmTime); err != nil { + if etag, err = CalculateEtag(n.ID, tmTime); err != nil { return } @@ -673,7 +668,7 @@ func (n *Node) AsResourceInfo(ctx context.Context, rp *provider.ResourcePermissi // use temporary etag if it is set if b, err := n.XattrString(ctx, prefixes.TmpEtagAttr); err == nil && b != "" { ri.Etag = fmt.Sprintf(`"%x"`, b) - } else if ri.Etag, err = calculateEtag(n, tmTime); err != nil { + } else if ri.Etag, err = CalculateEtag(n.ID, tmTime); err != nil { sublog.Debug().Err(err).Msg("could not calculate etag") } @@ -1150,6 +1145,18 @@ func (n *Node) DeleteGrant(ctx context.Context, g *provider.Grant, acquireLock b return nil } +// Purge removes a node from disk. It does not move it to the trash +func (n *Node) Purge(ctx context.Context) error { + // remove node + if err := utils.RemoveItem(n.InternalPath()); err != nil { + return err + } + + // remove child entry in parent + src := filepath.Join(n.ParentPath(), n.Name) + return os.Remove(src) +} + // ListGrants lists all grants of the current node. func (n *Node) ListGrants(ctx context.Context) ([]*provider.Grant, error) { grantees, err := n.ListGrantees(ctx) diff --git a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/revisions.go b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/revisions.go index e078d78ba..68022aa9f 100644 --- a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/revisions.go +++ b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/revisions.go @@ -91,7 +91,7 @@ func (fs *Decomposedfs) ListRevisions(ctx context.Context, ref *provider.Referen appctx.GetLogger(ctx).Error().Err(err).Str("name", fi.Name()).Msg("error reading blobsize xattr, using 0") } rev.Size = uint64(blobSize) - etag, err := node.CalculateEtag(n, mtime) + etag, err := node.CalculateEtag(n.ID, mtime) if err != nil { return nil, errors.Wrapf(err, "error calculating etag") } diff --git a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/spaces.go b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/spaces.go index d912cd9a1..a3745492d 100644 --- a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/spaces.go +++ b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/spaces.go @@ -964,7 +964,7 @@ func (fs *Decomposedfs) storageSpaceFromNode(ctx context.Context, n *node.Node, } } - etag, err := node.CalculateEtag(n, tmtime) + etag, err := node.CalculateEtag(n.ID, tmtime) if err != nil { return nil, err } diff --git a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload.go b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload.go index 64fd575d1..73dfd5496 100644 --- a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload.go +++ b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload.go @@ -23,13 +23,12 @@ import ( "fmt" "os" "path/filepath" - "regexp" "strings" "time" + "github.com/google/uuid" tusd "github.com/tus/tusd/pkg/handler" - userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" "github.com/cs3org/reva/v2/pkg/appctx" ctxpkg "github.com/cs3org/reva/v2/pkg/ctx" @@ -39,12 +38,11 @@ import ( "github.com/cs3org/reva/v2/pkg/storage/utils/chunking" "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node" "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload" + "github.com/cs3org/reva/v2/pkg/storagespace" "github.com/cs3org/reva/v2/pkg/utils" "github.com/pkg/errors" ) -var _idRegexp = regexp.MustCompile(".*/([^/]+).info") - // Upload uploads data to the given resource // TODO Upload (and InitiateUpload) needs a way to receive the expected checksum. // Maybe in metadata as 'checksum' => 'sha1 aeosvp45w5xaeoe' = lowercase, space separated? @@ -54,22 +52,19 @@ func (fs *Decomposedfs) Upload(ctx context.Context, req storage.UploadRequest, u return provider.ResourceInfo{}, errors.Wrap(err, "Decomposedfs: error retrieving upload") } - uploadInfo := up.(*upload.Upload) + session := up.(*upload.OcisSession) - p := uploadInfo.Info.Storage["NodeName"] - if chunking.IsChunked(p) { // check chunking v1 - var assembledFile string - p, assembledFile, err = fs.chunkHandler.WriteChunk(p, req.Body) + if session.Chunk() != "" { // check chunking v1 + p, assembledFile, err := fs.chunkHandler.WriteChunk(session.Chunk(), req.Body) if err != nil { return provider.ResourceInfo{}, err } if p == "" { - if err = uploadInfo.Terminate(ctx); err != nil { - return provider.ResourceInfo{}, errors.Wrap(err, "ocfs: error removing auxiliary files") + if err = session.Terminate(ctx); err != nil { + return provider.ResourceInfo{}, errors.Wrap(err, "Decomposedfs: error removing auxiliary files") } return provider.ResourceInfo{}, errtypes.PartialContent(req.Ref.String()) } - uploadInfo.Info.Storage["NodeName"] = p fd, err := os.Open(assembledFile) if err != nil { return provider.ResourceInfo{}, errors.Wrap(err, "Decomposedfs: error opening assembled file") @@ -79,46 +74,43 @@ func (fs *Decomposedfs) Upload(ctx context.Context, req storage.UploadRequest, u req.Body = fd } - if _, err := uploadInfo.WriteChunk(ctx, 0, req.Body); err != nil { + size, err := session.WriteChunk(ctx, 0, req.Body) + if err != nil { return provider.ResourceInfo{}, errors.Wrap(err, "Decomposedfs: error writing to binary file") } + session.SetSize(size) - if err := uploadInfo.FinishUpload(ctx); err != nil { + if err := session.FinishUpload(ctx); err != nil { return provider.ResourceInfo{}, err } if uff != nil { - info := uploadInfo.Info uploadRef := &provider.Reference{ ResourceId: &provider.ResourceId{ - StorageId: info.MetaData["providerID"], - SpaceId: info.Storage["SpaceRoot"], - OpaqueId: info.Storage["SpaceRoot"], + StorageId: session.ProviderID(), + SpaceId: session.SpaceID(), + OpaqueId: session.SpaceID(), }, - Path: utils.MakeRelativePath(filepath.Join(info.MetaData["dir"], info.MetaData["filename"])), + Path: utils.MakeRelativePath(filepath.Join(session.Dir(), session.Filename())), } - executant, ok := ctxpkg.ContextGetUser(uploadInfo.Ctx) - if !ok { - return provider.ResourceInfo{}, errtypes.PreconditionFailed("error getting user from uploadinfo context") - } - spaceOwner := &userpb.UserId{ - OpaqueId: info.Storage["SpaceOwnerOrManager"], - } - uff(spaceOwner, executant.Id, uploadRef) + executant := session.Executant() + uff(session.SpaceOwner(), &executant, uploadRef) } ri := provider.ResourceInfo{ // fill with at least fileid, mtime and etag Id: &provider.ResourceId{ - StorageId: uploadInfo.Info.MetaData["providerID"], - SpaceId: uploadInfo.Info.Storage["SpaceRoot"], - OpaqueId: uploadInfo.Info.Storage["NodeId"], + StorageId: session.ProviderID(), + SpaceId: session.SpaceID(), + OpaqueId: session.NodeID(), }, - Etag: uploadInfo.Info.MetaData["etag"], } - if mtime, err := utils.MTimeToTS(uploadInfo.Info.MetaData["mtime"]); err == nil { - ri.Mtime = &mtime + // add etag to metadata + ri.Etag, _ = node.CalculateEtag(session.NodeID(), session.MTime()) + + if !session.MTime().IsZero() { + ri.Mtime = utils.TimeToTS(session.MTime()) } return ri, nil @@ -130,6 +122,17 @@ func (fs *Decomposedfs) Upload(ctx context.Context, req storage.UploadRequest, u func (fs *Decomposedfs) InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error) { log := appctx.GetLogger(ctx) + // remember the path from the reference + refpath := ref.GetPath() + var chunk *chunking.ChunkBLOBInfo + var err error + if chunking.IsChunked(refpath) { // check chunking v1 + chunk, err = chunking.GetChunkBLOBInfo(refpath) + if err != nil { + return nil, errtypes.BadRequest(err.Error()) + } + ref.Path = chunk.Path + } n, err := fs.lu.NodeFromResource(ctx, ref) switch err.(type) { case nil: @@ -143,39 +146,44 @@ func (fs *Decomposedfs) InitiateUpload(ctx context.Context, ref *provider.Refere // permissions are checked in NewUpload below relative, err := fs.lu.Path(ctx, n, node.NoCheck) + // TODO why do we need the path here? + // jfd: it is used later when emitting the UploadReady event ... + // AAAND refPath might be . when accessing with an id / relative reference ... which causes NodeName to become . But then dir will also always be . + // That is why we still have to read the path here: so that the event we emit contains a relative reference with a path relative to the space root. WTF if err != nil { return nil, err } lockID, _ := ctxpkg.ContextGetLockID(ctx) - info := tusd.FileInfo{ - MetaData: tusd.MetaData{ - "filename": filepath.Base(relative), - "dir": filepath.Dir(relative), - "lockid": lockID, - }, - Size: uploadLength, - Storage: map[string]string{ - "SpaceRoot": n.SpaceRoot.ID, - "SpaceOwnerOrManager": n.SpaceOwnerOrManager(ctx).GetOpaqueId(), - }, + session := fs.sessionStore.New(ctx) + session.SetMetadata("filename", n.Name) + session.SetStorageValue("NodeName", n.Name) + if chunk != nil { + session.SetStorageValue("Chunk", filepath.Base(refpath)) } + session.SetMetadata("dir", filepath.Dir(relative)) + session.SetStorageValue("Dir", filepath.Dir(relative)) + session.SetMetadata("lockid", lockID) + + session.SetSize(uploadLength) + session.SetStorageValue("SpaceRoot", n.SpaceRoot.ID) // TODO SpaceRoot -> SpaceID + session.SetStorageValue("SpaceOwnerOrManager", n.SpaceOwnerOrManager(ctx).GetOpaqueId()) // TODO needed for what? if metadata != nil { - info.MetaData["providerID"] = metadata["providerID"] + session.SetMetadata("providerID", metadata["providerID"]) if mtime, ok := metadata["mtime"]; ok { if mtime != "null" { - info.MetaData["mtime"] = mtime + session.SetMetadata("mtime", metadata["mtime"]) } } if expiration, ok := metadata["expires"]; ok { if expiration != "null" { - info.MetaData["expires"] = expiration + session.SetMetadata("expires", metadata["expires"]) } } if _, ok := metadata["sizedeferred"]; ok { - info.SizeIsDeferred = true + session.SetSizeIsDeferred(true) } if checksum, ok := metadata["checksum"]; ok { parts := strings.SplitN(checksum, " ", 2) @@ -184,7 +192,7 @@ func (fs *Decomposedfs) InitiateUpload(ctx context.Context, ref *provider.Refere } switch parts[0] { case "sha1", "md5", "adler32": - info.MetaData["checksum"] = checksum + session.SetMetadata("checksum", checksum) default: return nil, errtypes.BadRequest("unsupported checksum algorithm: " + parts[0]) } @@ -192,35 +200,108 @@ func (fs *Decomposedfs) InitiateUpload(ctx context.Context, ref *provider.Refere // only check preconditions if they are not empty // TODO or is this a bad request? if metadata["if-match"] != "" { - info.MetaData["if-match"] = metadata["if-match"] + session.SetMetadata("if-match", metadata["if-match"]) } if metadata["if-none-match"] != "" { - info.MetaData["if-none-match"] = metadata["if-none-match"] + session.SetMetadata("if-none-match", metadata["if-none-match"]) } if metadata["if-unmodified-since"] != "" { - info.MetaData["if-unmodified-since"] = metadata["if-unmodified-since"] + session.SetMetadata("if-unmodified-since", metadata["if-unmodified-since"]) } } - log.Debug().Interface("info", info).Interface("node", n).Interface("metadata", metadata).Msg("Decomposedfs: resolved filename") + log.Debug().Interface("session", session).Interface("node", n).Interface("metadata", metadata).Msg("Decomposedfs: resolved filename") - _, err = node.CheckQuota(ctx, n.SpaceRoot, n.Exists, uint64(n.Blobsize), uint64(info.Size)) + _, err = node.CheckQuota(ctx, n.SpaceRoot, n.Exists, uint64(n.Blobsize), uint64(session.Size())) if err != nil { return nil, err } - upload, err := fs.NewUpload(ctx, info) + if session.Filename() == "" { + return nil, errors.New("Decomposedfs: missing filename in metadata") + } + if session.Dir() == "" { + return nil, errors.New("Decomposedfs: missing dir in metadata") + } + + // the parent owner will become the new owner + parent, perr := n.Parent(ctx) + if perr != nil { + return nil, errors.Wrap(perr, "Decomposedfs: error getting parent "+n.ParentID) + } + + // check permissions + var ( + checkNode *node.Node + path string + ) + if n.Exists { + // check permissions of file to be overwritten + checkNode = n + path, _ = storagespace.FormatReference(&provider.Reference{ResourceId: &provider.ResourceId{ + SpaceId: checkNode.SpaceID, + OpaqueId: checkNode.ID, + }}) + } else { + // check permissions of parent + checkNode = parent + path, _ = storagespace.FormatReference(&provider.Reference{ResourceId: &provider.ResourceId{ + SpaceId: checkNode.SpaceID, + OpaqueId: checkNode.ID, + }, Path: n.Name}) + } + rp, err := fs.p.AssemblePermissions(ctx, checkNode) + switch { + case err != nil: + return nil, err + case !rp.InitiateFileUpload: + return nil, errtypes.PermissionDenied(path) + } + + // are we trying to overwriting a folder with a file? + if n.Exists && n.IsDir(ctx) { + return nil, errtypes.PreconditionFailed("resource is not a file") + } + + // check lock + if err := n.CheckLock(ctx); err != nil { + return nil, err + } + + usr := ctxpkg.ContextMustGetUser(ctx) + + // fill future node info + if n.Exists { + if session.HeaderIfNoneMatch() == "*" { + return nil, errtypes.Aborted(fmt.Sprintf("parent %s already has a child %s, id %s", n.ParentID, n.Name, n.ID)) + } + session.SetStorageValue("NodeId", n.ID) + session.SetStorageValue("NodeExists", "true") + } else { + session.SetStorageValue("NodeId", uuid.New().String()) + } + session.SetStorageValue("NodeParentId", n.ParentID) + session.SetExecutant(usr) + session.SetStorageValue("LogLevel", log.GetLevel().String()) + + log.Debug().Interface("session", session).Msg("Decomposedfs: built session info") + // Create binary file in the upload folder with no content + // It will be used when determining the current offset of an upload + err = session.TouchBin() if err != nil { return nil, err } - info, _ = upload.GetInfo(ctx) + err = session.Persist(ctx) + if err != nil { + return nil, err + } metrics.UploadSessionsInitiated.Inc() return map[string]string{ - "simple": info.ID, - "tus": info.ID, + "simple": session.ID(), + "tus": session.ID(), }, nil } @@ -238,26 +319,26 @@ func (fs *Decomposedfs) UseIn(composer *tusd.StoreComposer) { // NewUpload returns a new tus Upload instance func (fs *Decomposedfs) NewUpload(ctx context.Context, info tusd.FileInfo) (tusd.Upload, error) { - return upload.New(ctx, info, fs.lu, fs.tp, fs.p, fs.o.Root, fs.stream, fs.o.AsyncFileUploads, fs.o.Tokens) + return nil, fmt.Errorf("not implemented, use InitiateUpload on the CS3 API to start a new upload") } // GetUpload returns the Upload for the given upload id func (fs *Decomposedfs) GetUpload(ctx context.Context, id string) (tusd.Upload, error) { - return upload.Get(ctx, id, fs.lu, fs.tp, fs.o.Root, fs.stream, fs.o.AsyncFileUploads, fs.o.Tokens) + return fs.sessionStore.Get(ctx, id) } // ListUploadSessions returns the upload sessions for the given filter func (fs *Decomposedfs) ListUploadSessions(ctx context.Context, filter storage.UploadSessionFilter) ([]storage.UploadSession, error) { - var sessions []storage.UploadSession + var sessions []*upload.OcisSession if filter.ID != nil && *filter.ID != "" { - session, err := fs.getUploadSession(ctx, filepath.Join(fs.o.Root, "uploads", *filter.ID+".info")) + session, err := fs.sessionStore.Get(ctx, *filter.ID) if err != nil { return nil, err } - sessions = []storage.UploadSession{session} + sessions = []*upload.OcisSession{session} } else { var err error - sessions, err = fs.uploadSessions(ctx) + sessions, err = fs.sessionStore.List(ctx) if err != nil { return nil, err } @@ -288,64 +369,19 @@ func (fs *Decomposedfs) ListUploadSessions(ctx context.Context, filter storage.U // To implement the termination extension as specified in https://tus.io/protocols/resumable-upload.html#termination // the storage needs to implement AsTerminatableUpload func (fs *Decomposedfs) AsTerminatableUpload(up tusd.Upload) tusd.TerminatableUpload { - return up.(*upload.Upload) + return up.(*upload.OcisSession) } // AsLengthDeclarableUpload returns a LengthDeclarableUpload // To implement the creation-defer-length extension as specified in https://tus.io/protocols/resumable-upload.html#creation // the storage needs to implement AsLengthDeclarableUpload func (fs *Decomposedfs) AsLengthDeclarableUpload(up tusd.Upload) tusd.LengthDeclarableUpload { - return up.(*upload.Upload) + return up.(*upload.OcisSession) } // AsConcatableUpload returns a ConcatableUpload // To implement the concatenation extension as specified in https://tus.io/protocols/resumable-upload.html#concatenation // the storage needs to implement AsConcatableUpload func (fs *Decomposedfs) AsConcatableUpload(up tusd.Upload) tusd.ConcatableUpload { - return up.(*upload.Upload) -} - -func (fs *Decomposedfs) uploadSessions(ctx context.Context) ([]storage.UploadSession, error) { - uploads := []storage.UploadSession{} - infoFiles, err := filepath.Glob(filepath.Join(fs.o.Root, "uploads", "*.info")) - if err != nil { - return nil, err - } - - for _, info := range infoFiles { - progress, err := fs.getUploadSession(ctx, info) - if err != nil { - appctx.GetLogger(ctx).Error().Interface("path", info).Msg("Decomposedfs: could not getUploadSession") - continue - } - - uploads = append(uploads, progress) - } - return uploads, nil -} - -func (fs *Decomposedfs) getUploadSession(ctx context.Context, path string) (storage.UploadSession, error) { - match := _idRegexp.FindStringSubmatch(path) - if match == nil || len(match) < 2 { - return nil, fmt.Errorf("invalid upload path") - } - up, err := fs.GetUpload(ctx, match[1]) - if err != nil { - return nil, err - } - info, err := up.GetInfo(context.Background()) - if err != nil { - return nil, err - } - // upload processing state is stored in the node, for decomposedfs the NodeId is always set by InitiateUpload - n, err := node.ReadNode(ctx, fs.lu, info.Storage["SpaceRoot"], info.Storage["NodeId"], true, nil, true) - if err != nil { - return nil, err - } - progress := upload.Progress{ - Path: path, - Info: info, - Processing: n.IsProcessing(ctx), - } - return progress, nil + return up.(*upload.OcisSession) } diff --git a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/processing.go b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/processing.go deleted file mode 100644 index a025c54ed..000000000 --- a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/processing.go +++ /dev/null @@ -1,582 +0,0 @@ -// Copyright 2018-2022 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 upload - -import ( - "context" - "encoding/json" - stderrors "errors" - "fmt" - iofs "io/fs" - "os" - "path/filepath" - "strconv" - "strings" - "time" - - userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" - provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" - "github.com/cs3org/reva/v2/pkg/appctx" - ctxpkg "github.com/cs3org/reva/v2/pkg/ctx" - "github.com/cs3org/reva/v2/pkg/errtypes" - "github.com/cs3org/reva/v2/pkg/events" - "github.com/cs3org/reva/v2/pkg/logger" - "github.com/cs3org/reva/v2/pkg/storage/utils/chunking" - "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/lookup" - "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata/prefixes" - "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node" - "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/options" - "github.com/cs3org/reva/v2/pkg/storagespace" - "github.com/cs3org/reva/v2/pkg/utils" - "github.com/google/uuid" - "github.com/pkg/errors" - "github.com/rogpeppe/go-internal/lockedfile" - tusd "github.com/tus/tusd/pkg/handler" -) - -var defaultFilePerm = os.FileMode(0664) - -// PermissionsChecker defines an interface for checking permissions on a Node -type PermissionsChecker interface { - AssemblePermissions(ctx context.Context, n *node.Node) (ap provider.ResourcePermissions, err error) -} - -// New returns a new processing instance -func New(ctx context.Context, info tusd.FileInfo, lu *lookup.Lookup, tp Tree, p PermissionsChecker, fsRoot string, pub events.Publisher, async bool, tknopts options.TokenOptions) (upload *Upload, err error) { - - log := appctx.GetLogger(ctx) - log.Debug().Interface("info", info).Msg("Decomposedfs: NewUpload") - - if info.MetaData["filename"] == "" { - return nil, errors.New("Decomposedfs: missing filename in metadata") - } - if info.MetaData["dir"] == "" { - return nil, errors.New("Decomposedfs: missing dir in metadata") - } - - n, err := lu.NodeFromSpaceID(ctx, info.Storage["SpaceRoot"]) - if err != nil { - return nil, errors.Wrap(err, "Decomposedfs: error getting space root node") - } - - n, err = lookupNode(ctx, n, filepath.Join(info.MetaData["dir"], info.MetaData["filename"]), lu) - if err != nil { - return nil, errors.Wrap(err, "Decomposedfs: error walking path") - } - - log.Debug().Interface("info", info).Interface("node", n).Msg("Decomposedfs: resolved filename") - - // the parent owner will become the new owner - parent, perr := n.Parent(ctx) - if perr != nil { - return nil, errors.Wrap(perr, "Decomposedfs: error getting parent "+n.ParentID) - } - - // check permissions - var ( - checkNode *node.Node - path string - ) - if n.Exists { - // check permissions of file to be overwritten - checkNode = n - path, _ = storagespace.FormatReference(&provider.Reference{ResourceId: &provider.ResourceId{ - SpaceId: checkNode.SpaceID, - OpaqueId: checkNode.ID, - }}) - } else { - // check permissions of parent - checkNode = parent - path, _ = storagespace.FormatReference(&provider.Reference{ResourceId: &provider.ResourceId{ - SpaceId: checkNode.SpaceID, - OpaqueId: checkNode.ID, - }, Path: n.Name}) - } - rp, err := p.AssemblePermissions(ctx, checkNode) - switch { - case err != nil: - return nil, err - case !rp.InitiateFileUpload: - return nil, errtypes.PermissionDenied(path) - } - - // are we trying to overwriting a folder with a file? - if n.Exists && n.IsDir(ctx) { - return nil, errtypes.PreconditionFailed("resource is not a file") - } - - // check lock - if info.MetaData["lockid"] != "" { - ctx = ctxpkg.ContextSetLockID(ctx, info.MetaData["lockid"]) - } - if err := n.CheckLock(ctx); err != nil { - return nil, err - } - - info.ID = uuid.New().String() - - binPath := filepath.Join(fsRoot, "uploads", info.ID) - usr := ctxpkg.ContextMustGetUser(ctx) - - var ( - spaceRoot string - ok bool - ) - if info.Storage != nil { - if spaceRoot, ok = info.Storage["SpaceRoot"]; !ok { - spaceRoot = n.SpaceRoot.ID - } - } else { - spaceRoot = n.SpaceRoot.ID - } - - info.Storage = map[string]string{ - "Type": "OCISStore", - "BinPath": binPath, - - "NodeId": n.ID, - "NodeExists": "true", - "NodeParentId": n.ParentID, - "NodeName": n.Name, - "SpaceRoot": spaceRoot, - "SpaceOwnerOrManager": info.Storage["SpaceOwnerOrManager"], - - "Idp": usr.Id.Idp, - "UserId": usr.Id.OpaqueId, - "UserType": utils.UserTypeToString(usr.Id.Type), - "UserName": usr.Username, - - "LogLevel": log.GetLevel().String(), - } - if !n.Exists { - // fill future node info - info.Storage["NodeId"] = uuid.New().String() - info.Storage["NodeExists"] = "false" - } - if info.MetaData["if-none-match"] == "*" && info.Storage["NodeExists"] == "true" { - return nil, errtypes.Aborted(fmt.Sprintf("parent %s already has a child %s", n.ID, n.Name)) - } - // Create binary file in the upload folder with no content - log.Debug().Interface("info", info).Msg("Decomposedfs: built storage info") - file, err := os.OpenFile(binPath, os.O_CREATE|os.O_WRONLY, defaultFilePerm) - if err != nil { - return nil, err - } - defer file.Close() - - u := buildUpload(ctx, info, binPath, filepath.Join(fsRoot, "uploads", info.ID+".info"), lu, tp, pub, async, tknopts) - - // writeInfo creates the file by itself if necessary - err = u.writeInfo() - if err != nil { - return nil, err - } - - return u, nil -} - -// Get returns the Upload for the given upload id -func Get(ctx context.Context, id string, lu *lookup.Lookup, tp Tree, fsRoot string, pub events.Publisher, async bool, tknopts options.TokenOptions) (*Upload, error) { - infoPath := filepath.Join(fsRoot, "uploads", id+".info") - - info := tusd.FileInfo{} - data, err := os.ReadFile(infoPath) - if err != nil { - if errors.Is(err, iofs.ErrNotExist) { - // Interpret os.ErrNotExist as 404 Not Found - err = tusd.ErrNotFound - } - return nil, err - } - if err := json.Unmarshal(data, &info); err != nil { - return nil, err - } - - stat, err := os.Stat(info.Storage["BinPath"]) - if err != nil { - return nil, err - } - - info.Offset = stat.Size() - - u := &userpb.User{ - Id: &userpb.UserId{ - Idp: info.Storage["Idp"], - OpaqueId: info.Storage["UserId"], - Type: utils.UserTypeMap(info.Storage["UserType"]), - }, - Username: info.Storage["UserName"], - } - - ctx = ctxpkg.ContextSetUser(ctx, u) - - // restore logger from file info - log, err := logger.FromConfig(&logger.LogConf{ - Output: "stderr", // TODO use config from decomposedfs - Mode: "json", // TODO use config from decomposedfs - Level: info.Storage["LogLevel"], - }) - if err != nil { - return nil, err - } - sub := log.With().Int("pid", os.Getpid()).Logger() - ctx = appctx.WithLogger(ctx, &sub) - - // TODO store and add traceid in file info - - up := buildUpload(ctx, info, info.Storage["BinPath"], infoPath, lu, tp, pub, async, tknopts) - up.versionsPath = info.MetaData["versionsPath"] - up.SizeDiff, _ = strconv.ParseInt(info.MetaData["sizeDiff"], 10, 64) - return up, nil -} - -// CreateNodeForUpload will create the target node for the Upload -func CreateNodeForUpload(upload *Upload, initAttrs node.Attributes) (*node.Node, error) { - ctx, span := tracer.Start(upload.Ctx, "CreateNodeForUpload") - defer span.End() - _, subspan := tracer.Start(ctx, "os.Stat") - fi, err := os.Stat(upload.binPath) - subspan.End() - if err != nil { - return nil, err - } - - fsize := fi.Size() - spaceID := upload.Info.Storage["SpaceRoot"] - n := node.New( - spaceID, - upload.Info.Storage["NodeId"], - upload.Info.Storage["NodeParentId"], - upload.Info.Storage["NodeName"], - fsize, - upload.Info.ID, - provider.ResourceType_RESOURCE_TYPE_FILE, - nil, - upload.lu, - ) - n.SpaceRoot, err = node.ReadNode(ctx, upload.lu, spaceID, spaceID, false, nil, false) - if err != nil { - return nil, err - } - - // check lock - if err := n.CheckLock(ctx); err != nil { - return nil, err - } - - var f *lockedfile.File - switch upload.Info.Storage["NodeExists"] { - case "false": - f, err = initNewNode(upload, n, uint64(fsize)) - if f != nil { - appctx.GetLogger(upload.Ctx).Info().Str("lockfile", f.Name()).Interface("err", err).Msg("got lock file from initNewNode") - } - default: - f, err = updateExistingNode(upload, n, spaceID, uint64(fsize)) - if f != nil { - appctx.GetLogger(upload.Ctx).Info().Str("lockfile", f.Name()).Interface("err", err).Msg("got lock file from updateExistingNode") - } - } - defer func() { - if f == nil { - return - } - if err := f.Close(); err != nil { - appctx.GetLogger(upload.Ctx).Error().Err(err).Str("nodeid", n.ID).Str("parentid", n.ParentID).Msg("could not close lock") - } - }() - if err != nil { - return nil, err - } - - mtime := time.Now() - if upload.Info.MetaData["mtime"] != "" { - // overwrite mtime if requested - mtime, err = utils.MTimeToTime(upload.Info.MetaData["mtime"]) - if err != nil { - return nil, err - } - } - - // overwrite technical information - initAttrs.SetString(prefixes.MTimeAttr, mtime.UTC().Format(time.RFC3339Nano)) - initAttrs.SetInt64(prefixes.TypeAttr, int64(provider.ResourceType_RESOURCE_TYPE_FILE)) - initAttrs.SetString(prefixes.ParentidAttr, n.ParentID) - initAttrs.SetString(prefixes.NameAttr, n.Name) - initAttrs.SetString(prefixes.BlobIDAttr, n.BlobID) - initAttrs.SetInt64(prefixes.BlobsizeAttr, n.Blobsize) - initAttrs.SetString(prefixes.StatusPrefix, node.ProcessingStatus+upload.Info.ID) - - // update node metadata with new blobid etc - err = n.SetXattrsWithContext(ctx, initAttrs, false) - if err != nil { - return nil, errors.Wrap(err, "Decomposedfs: could not write metadata") - } - - // add etag to metadata - upload.Info.MetaData["etag"], _ = node.CalculateEtag(n, mtime) - - // update nodeid for later - upload.Info.Storage["NodeId"] = n.ID - if err := upload.writeInfo(); err != nil { - return nil, err - } - - return n, nil -} - -func initNewNode(upload *Upload, n *node.Node, fsize uint64) (*lockedfile.File, error) { - // create folder structure (if needed) - if err := os.MkdirAll(filepath.Dir(n.InternalPath()), 0700); err != nil { - return nil, err - } - - // create and write lock new node metadata - f, err := lockedfile.OpenFile(upload.lu.MetadataBackend().LockfilePath(n.InternalPath()), os.O_RDWR|os.O_CREATE, 0600) - if err != nil { - return nil, err - } - - // we also need to touch the actual node file here it stores the mtime of the resource - h, err := os.OpenFile(n.InternalPath(), os.O_CREATE|os.O_EXCL, 0600) - if err != nil { - return f, err - } - h.Close() - - if _, err := node.CheckQuota(upload.Ctx, n.SpaceRoot, false, 0, fsize); err != nil { - return f, err - } - - // link child name to parent if it is new - childNameLink := filepath.Join(n.ParentPath(), n.Name) - relativeNodePath := filepath.Join("../../../../../", lookup.Pathify(n.ID, 4, 2)) - log := appctx.GetLogger(upload.Ctx).With().Str("childNameLink", childNameLink).Str("relativeNodePath", relativeNodePath).Logger() - log.Info().Msg("initNewNode: creating symlink") - - if err = os.Symlink(relativeNodePath, childNameLink); err != nil { - log.Info().Err(err).Msg("initNewNode: symlink failed") - if errors.Is(err, iofs.ErrExist) { - log.Info().Err(err).Msg("initNewNode: symlink already exists") - return f, errtypes.AlreadyExists(n.Name) - } - return f, errors.Wrap(err, "Decomposedfs: could not symlink child entry") - } - log.Info().Msg("initNewNode: symlink created") - - // on a new file the sizeDiff is the fileSize - upload.SizeDiff = int64(fsize) - upload.Info.MetaData["sizeDiff"] = strconv.Itoa(int(upload.SizeDiff)) - return f, nil -} - -func updateExistingNode(upload *Upload, n *node.Node, spaceID string, fsize uint64) (*lockedfile.File, error) { - targetPath := n.InternalPath() - - // write lock existing node before reading any metadata - f, err := lockedfile.OpenFile(upload.lu.MetadataBackend().LockfilePath(targetPath), os.O_RDWR|os.O_CREATE, 0600) - if err != nil { - return nil, err - } - - old, _ := node.ReadNode(upload.Ctx, upload.lu, spaceID, n.ID, false, nil, false) - if _, err := node.CheckQuota(upload.Ctx, n.SpaceRoot, true, uint64(old.Blobsize), fsize); err != nil { - return f, err - } - - oldNodeMtime, err := old.GetMTime(upload.Ctx) - if err != nil { - return f, err - } - oldNodeEtag, err := node.CalculateEtag(old, oldNodeMtime) - if err != nil { - return f, err - } - - // When the if-match header was set we need to check if the - // etag still matches before finishing the upload. - if ifMatch, ok := upload.Info.MetaData["if-match"]; ok { - if ifMatch != oldNodeEtag { - return f, errtypes.Aborted("etag mismatch") - } - } - - // When the if-none-match header was set we need to check if any of the - // etags matches before finishing the upload. - if ifNoneMatch, ok := upload.Info.MetaData["if-none-match"]; ok { - if ifNoneMatch == "*" { - return f, errtypes.Aborted("etag mismatch, resource exists") - } - for _, ifNoneMatchTag := range strings.Split(ifNoneMatch, ",") { - if ifNoneMatchTag == oldNodeEtag { - return f, errtypes.Aborted("etag mismatch") - } - } - } - - // When the if-unmodified-since header was set we need to check if the - // etag still matches before finishing the upload. - if ifUnmodifiedSince, ok := upload.Info.MetaData["if-unmodified-since"]; ok { - if err != nil { - return f, errtypes.InternalError(fmt.Sprintf("failed to read mtime of node: %s", err)) - } - ifUnmodifiedSince, err := time.Parse(time.RFC3339Nano, ifUnmodifiedSince) - if err != nil { - return f, errtypes.InternalError(fmt.Sprintf("failed to parse if-unmodified-since time: %s", err)) - } - - if oldNodeMtime.After(ifUnmodifiedSince) { - return f, errtypes.Aborted("if-unmodified-since mismatch") - } - } - - upload.versionsPath = upload.lu.InternalPath(spaceID, n.ID+node.RevisionIDDelimiter+oldNodeMtime.UTC().Format(time.RFC3339Nano)) - upload.SizeDiff = int64(fsize) - old.Blobsize - upload.Info.MetaData["versionsPath"] = upload.versionsPath - upload.Info.MetaData["sizeDiff"] = strconv.Itoa(int(upload.SizeDiff)) - - // create version node - if _, err := os.Create(upload.versionsPath); err != nil { - return f, err - } - - // copy blob metadata to version node - if err := upload.lu.CopyMetadataWithSourceLock(upload.Ctx, targetPath, upload.versionsPath, func(attributeName string, value []byte) (newValue []byte, copy bool) { - return value, strings.HasPrefix(attributeName, prefixes.ChecksumPrefix) || - attributeName == prefixes.TypeAttr || - attributeName == prefixes.BlobIDAttr || - attributeName == prefixes.BlobsizeAttr || - attributeName == prefixes.MTimeAttr - }, f, true); err != nil { - return f, err - } - - // keep mtime from previous version - if err := os.Chtimes(upload.versionsPath, oldNodeMtime, oldNodeMtime); err != nil { - return f, errtypes.InternalError(fmt.Sprintf("failed to change mtime of version node: %s", err)) - } - - return f, nil -} - -// lookupNode looks up nodes by path. -// This method can also handle lookups for paths which contain chunking information. -func lookupNode(ctx context.Context, spaceRoot *node.Node, path string, lu *lookup.Lookup) (*node.Node, error) { - p := path - isChunked := chunking.IsChunked(path) - if isChunked { - chunkInfo, err := chunking.GetChunkBLOBInfo(path) - if err != nil { - return nil, err - } - p = chunkInfo.Path - } - - n, err := lu.WalkPath(ctx, spaceRoot, p, true, func(ctx context.Context, n *node.Node) error { return nil }) - if err != nil { - return nil, errors.Wrap(err, "Decomposedfs: error walking path") - } - - if isChunked { - n.Name = filepath.Base(path) - } - return n, nil -} - -// Progress adapts the persisted upload metadata for the UploadSessionLister interface -type Progress struct { - Path string - Info tusd.FileInfo - Processing bool -} - -// ID implements the storage.UploadSession interface -func (p Progress) ID() string { - return p.Info.ID -} - -// Filename implements the storage.UploadSession interface -func (p Progress) Filename() string { - return p.Info.MetaData["filename"] -} - -// Size implements the storage.UploadSession interface -func (p Progress) Size() int64 { - return p.Info.Size -} - -// Offset implements the storage.UploadSession interface -func (p Progress) Offset() int64 { - return p.Info.Offset -} - -// Reference implements the storage.UploadSession interface -func (p Progress) Reference() provider.Reference { - return provider.Reference{ - ResourceId: &provider.ResourceId{ - StorageId: p.Info.MetaData["providerID"], - SpaceId: p.Info.Storage["SpaceRoot"], - OpaqueId: p.Info.Storage["NodeId"], // Node id is always set in InitiateUpload - }, - } -} - -// Executant implements the storage.UploadSession interface -func (p Progress) Executant() userpb.UserId { - return userpb.UserId{ - Idp: p.Info.Storage["Idp"], - OpaqueId: p.Info.Storage["UserId"], - Type: utils.UserTypeMap(p.Info.Storage["UserType"]), - } -} - -// SpaceOwner implements the storage.UploadSession interface -func (p Progress) SpaceOwner() *userpb.UserId { - return &userpb.UserId{ - // idp and type do not seem to be consumed and the node currently only stores the user id anyway - OpaqueId: p.Info.Storage["SpaceOwnerOrManager"], - } -} - -// Expires implements the storage.UploadSession interface -func (p Progress) Expires() time.Time { - mt, _ := utils.MTimeToTime(p.Info.MetaData["expires"]) - return mt -} - -// IsProcessing implements the storage.UploadSession interface -func (p Progress) IsProcessing() bool { - return p.Processing -} - -// Purge implements the storage.UploadSession interface -func (p Progress) Purge(ctx context.Context) error { - berr := os.Remove(p.Info.Storage["BinPath"]) - if berr != nil { - appctx.GetLogger(ctx).Error().Str("id", p.Info.ID).Interface("path", p.Info.Storage["BinPath"]).Msg("Decomposedfs: could not purge bin path for upload session") - } - - // remove upload metadata - merr := os.Remove(p.Path) - if merr != nil { - appctx.GetLogger(ctx).Error().Str("id", p.Info.ID).Interface("path", p.Path).Msg("Decomposedfs: could not purge metadata path for upload session") - } - - return stderrors.Join(berr, merr) -} diff --git a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/session.go b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/session.go new file mode 100644 index 000000000..bd9dc085f --- /dev/null +++ b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/session.go @@ -0,0 +1,319 @@ +// Copyright 2018-2023 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 upload + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "strconv" + "time" + + tusd "github.com/tus/tusd/pkg/handler" + + userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1" + provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" + "github.com/cs3org/reva/v2/pkg/appctx" + ctxpkg "github.com/cs3org/reva/v2/pkg/ctx" + "github.com/cs3org/reva/v2/pkg/logger" + "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node" + "github.com/cs3org/reva/v2/pkg/utils" +) + +// OcisSession extends tus upload lifecycle with postprocessing steps. +type OcisSession struct { + store OcisStore + // for now, we keep the json files in the uploads folder + info tusd.FileInfo +} + +// Context returns a context with the user, logger and lockid used when initiating the upload session +func (s *OcisSession) Context(ctx context.Context) context.Context { // restore logger from file info + log, _ := logger.FromConfig(&logger.LogConf{ + Output: "stderr", // TODO use config from decomposedfs + Mode: "json", // TODO use config from decomposedfs + Level: s.info.Storage["LogLevel"], + }) + sub := log.With().Int("pid", os.Getpid()).Logger() + ctx = appctx.WithLogger(ctx, &sub) + ctx = ctxpkg.ContextSetLockID(ctx, s.lockID()) + return ctxpkg.ContextSetUser(ctx, s.executantUser()) +} + +func (s *OcisSession) lockID() string { + return s.info.MetaData["lockid"] +} +func (s *OcisSession) executantUser() *userpb.User { + return &userpb.User{ + Id: &userpb.UserId{ + Type: userpb.UserType(userpb.UserType_value[s.info.Storage["UserType"]]), + Idp: s.info.Storage["Idp"], + OpaqueId: s.info.Storage["UserId"], + }, + Username: s.info.Storage["UserName"], + } +} + +// Purge deletes the upload session metadata and written binary data +func (s *OcisSession) Purge(ctx context.Context) error { + if err := os.Remove(sessionPath(s.store.root, s.info.ID)); err != nil { + return err + } + if err := os.Remove(s.binPath()); err != nil { + return err + } + return nil +} + +// TouchBin creates a file to contain the binary data. It's size will be used to keep track of the tus upload offset. +func (s *OcisSession) TouchBin() error { + file, err := os.OpenFile(s.binPath(), os.O_CREATE|os.O_WRONLY, defaultFilePerm) + if err != nil { + return err + } + return file.Close() +} + +// Persist writes the upload session metadata to disk +func (s *OcisSession) Persist(ctx context.Context) error { + uploadPath := sessionPath(s.store.root, s.info.ID) + // create folder structure (if needed) + if err := os.MkdirAll(filepath.Dir(uploadPath), 0700); err != nil { + return err + } + + var d []byte + d, err := json.Marshal(s.info) + if err != nil { + return err + } + + return os.WriteFile(uploadPath, d, 0600) +} + +// ToFileInfo returns tus compatible FileInfo so the tus handler can access the upload offset +func (s *OcisSession) ToFileInfo() tusd.FileInfo { + return s.info +} + +// ProviderID returns the provider id +func (s *OcisSession) ProviderID() string { + return s.info.MetaData["providerID"] +} + +// SpaceID returns the space id +func (s *OcisSession) SpaceID() string { + return s.info.Storage["SpaceRoot"] +} + +// NodeID returns the node id +func (s *OcisSession) NodeID() string { + return s.info.Storage["NodeId"] +} + +// NodeParentID returns the nodes parent id +func (s *OcisSession) NodeParentID() string { + return s.info.Storage["NodeParentId"] +} + +// NodeExists returns wether or not the node existed during InitiateUpload. +// FIXME If two requests try to write the same file they both will store a new +// random node id in the session and try to initialize a new node when +// finishing the upload. The second request will fail with an already exists +// error when trying to create the symlink for the node in the parent directory. +// A node should be created as part of InitiateUpload. When listing a directory +// we can decide if we want to skip the entry, or expose uploed progress +// information. But that is a bigger change and might involve client work. +func (s *OcisSession) NodeExists() bool { + return s.info.Storage["NodeExists"] == "true" +} + +// HeaderIfMatch returns the if-match header for the upload session +func (s *OcisSession) HeaderIfMatch() string { + return s.info.MetaData["if-match"] +} + +// HeaderIfNoneMatch returns the if-none-match header for the upload session +func (s *OcisSession) HeaderIfNoneMatch() string { + return s.info.MetaData["if-none-match"] +} + +// HeaderIfUnmodifiedSince returns the if-unmodified-since header for the upload session +func (s *OcisSession) HeaderIfUnmodifiedSince() string { + return s.info.MetaData["if-unmodified-since"] +} + +// Node returns the node for the session +func (s *OcisSession) Node(ctx context.Context) (*node.Node, error) { + n, err := node.ReadNode(ctx, s.store.lu, s.SpaceID(), s.info.Storage["NodeId"], false, nil, true) + if err != nil { + return nil, err + } + return n, nil +} + +// ID returns the upload session id +func (s *OcisSession) ID() string { + return s.info.ID +} + +// Filename returns the name of the node which is not the same as the name af the file being uploaded for legacy chunked uploads +func (s *OcisSession) Filename() string { + return s.info.Storage["NodeName"] +} + +// Chunk returns the chunk name when a legacy chunked upload was started +func (s *OcisSession) Chunk() string { + return s.info.Storage["Chunk"] +} + +// SetMetadata is used to fill the upload metadata that will be exposed to the end user +func (s *OcisSession) SetMetadata(key, value string) { + s.info.MetaData[key] = value +} + +// SetStorageValue is used to set metadata only relevant for the upload session implementation +func (s *OcisSession) SetStorageValue(key, value string) { + s.info.Storage[key] = value +} + +// SetSize will set the upload size of the underlying tus info. +func (s *OcisSession) SetSize(size int64) { + s.info.Size = size +} + +// SetSizeIsDeferred is uset to change the SizeIsDeferred property of the underlying tus info. +func (s *OcisSession) SetSizeIsDeferred(value bool) { + s.info.SizeIsDeferred = value +} + +// Dir returns the directory to which the upload is made +// TODO get rid of Dir(), whoever consumes the reference should be able to deal +// with a relative reference. +// Dir is only used to: +// - fill the Path property when emitting the UploadReady event after +// postprocessing finished. I wonder why the UploadReady contains a finished +// flag ... maybe multiple distinct events would make more sense. +// - build the reference that is passed to the FileUploaded event in the +// UploadFinishedFunc callback passed to the Upload call used for simple +// datatx put requests +// +// AFAICT only search and audit services consume the path. +// - search needs to index from the root anyway. And it only needs the most +// recent path to put it in the index. So it should already be able to deal +// with an id based reference. +// - audit on the other hand needs to log events with the path at the state of +// the event ... so it does need the full path. +// +// I think we can safely determine the path later, right before emitting the +// event. And maybe make it configurable, because only audit needs it, anyway. +func (s *OcisSession) Dir() string { + return s.info.Storage["Dir"] +} + +// Size returns the upload size +func (s *OcisSession) Size() int64 { + return s.info.Size +} + +// SizeDiff returns the size diff that was calculated after postprocessing +func (s *OcisSession) SizeDiff() int64 { + sizeDiff, _ := strconv.ParseInt(s.info.MetaData["sizeDiff"], 10, 64) + return sizeDiff +} + +// Reference returns a reference that can be used to access the uploaded resource +func (s *OcisSession) Reference() provider.Reference { + return provider.Reference{ + ResourceId: &provider.ResourceId{ + StorageId: s.info.MetaData["providerID"], + SpaceId: s.info.Storage["SpaceRoot"], + OpaqueId: s.info.Storage["NodeId"], + }, + // Path is not used + } +} + +// Executant returns the id of the user that initiated the upload session +func (s *OcisSession) Executant() userpb.UserId { + return userpb.UserId{ + Type: userpb.UserType(userpb.UserType_value[s.info.Storage["UserType"]]), + Idp: s.info.Storage["Idp"], + OpaqueId: s.info.Storage["UserId"], + } +} + +// SetExecutant is used to remember the user that initiated the upload session +func (s *OcisSession) SetExecutant(u *userpb.User) { + s.info.Storage["Idp"] = u.GetId().GetIdp() + s.info.Storage["UserId"] = u.GetId().GetOpaqueId() + s.info.Storage["UserType"] = utils.UserTypeToString(u.GetId().Type) + s.info.Storage["UserName"] = u.GetUsername() +} + +// Offset returns the current upload offset +func (s *OcisSession) Offset() int64 { + return s.info.Offset +} + +// SpaceOwner returns the id of the space owner +func (s *OcisSession) SpaceOwner() *userpb.UserId { + return &userpb.UserId{ + // idp and type do not seem to be consumed and the node currently only stores the user id anyway + OpaqueId: s.info.Storage["SpaceOwnerOrManager"], + } +} + +// Expires returns the time the upload session expires +func (s *OcisSession) Expires() time.Time { + var t time.Time + if value, ok := s.info.MetaData["expires"]; ok { + t, _ = utils.MTimeToTime(value) + } + return t +} + +// MTime returns the mtime to use for the uploaded file +func (s *OcisSession) MTime() time.Time { + var t time.Time + if value, ok := s.info.MetaData["mtime"]; ok { + t, _ = utils.MTimeToTime(value) + } + return t +} + +// IsProcessing returns true if the node has entered postprocessing state +func (s *OcisSession) IsProcessing() bool { + n, err := s.Node(context.Background()) + if err != nil { + return false + } + return n.IsProcessing(context.Background()) +} + +// binPath returns the path to the file storing the binary data. +func (s *OcisSession) binPath() string { + return filepath.Join(s.store.root, "uploads", s.info.ID) +} + +// sessionPath returns the path to the .info file storing the file's info. +func sessionPath(root, id string) string { + return filepath.Join(root, "uploads", id+".info") +} diff --git a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/store.go b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/store.go new file mode 100644 index 000000000..fc6bd4708 --- /dev/null +++ b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/store.go @@ -0,0 +1,380 @@ +// Copyright 2018-2022 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 upload + +import ( + "context" + "encoding/json" + "fmt" + iofs "io/fs" + "os" + "path/filepath" + "regexp" + "strconv" + "strings" + "time" + + provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1" + "github.com/cs3org/reva/v2/pkg/appctx" + "github.com/cs3org/reva/v2/pkg/errtypes" + "github.com/cs3org/reva/v2/pkg/events" + "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/lookup" + "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata/prefixes" + "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node" + "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/options" + "github.com/google/uuid" + "github.com/pkg/errors" + "github.com/rogpeppe/go-internal/lockedfile" + tusd "github.com/tus/tusd/pkg/handler" +) + +var _idRegexp = regexp.MustCompile(".*/([^/]+).info") + +// PermissionsChecker defines an interface for checking permissions on a Node +type PermissionsChecker interface { + AssemblePermissions(ctx context.Context, n *node.Node) (ap provider.ResourcePermissions, err error) +} + +// OcisStore manages upload sessions +type OcisStore struct { + lu *lookup.Lookup + tp Tree + root string + pub events.Publisher + async bool + tknopts options.TokenOptions +} + +// NewSessionStore returns a new OcisStore +func NewSessionStore(lu *lookup.Lookup, tp Tree, root string, pub events.Publisher, async bool, tknopts options.TokenOptions) *OcisStore { + return &OcisStore{ + lu: lu, + tp: tp, + root: root, + pub: pub, + async: async, + tknopts: tknopts, + } +} + +// New returns a new upload session +func (store OcisStore) New(ctx context.Context) *OcisSession { + return &OcisSession{ + store: store, + info: tusd.FileInfo{ + ID: uuid.New().String(), + Storage: map[string]string{ + "Type": "OCISStore", + }, + MetaData: tusd.MetaData{}, + }, + } +} + +// List lists all upload sessions +func (store OcisStore) List(ctx context.Context) ([]*OcisSession, error) { + uploads := []*OcisSession{} + infoFiles, err := filepath.Glob(filepath.Join(store.root, "uploads", "*.info")) + if err != nil { + return nil, err + } + + for _, info := range infoFiles { + progress, err := store.Get(ctx, info) + if err != nil { + appctx.GetLogger(ctx).Error().Interface("path", info).Msg("Decomposedfs: could not getUploadSession") + continue + } + + uploads = append(uploads, progress) + } + return uploads, nil +} + +// Get returns the upload session for the given upload id +func (store OcisStore) Get(ctx context.Context, id string) (*OcisSession, error) { + sessionPath := filepath.Join(store.root, "uploads", id+".info") + match := _idRegexp.FindStringSubmatch(sessionPath) + if match == nil || len(match) < 2 { + return nil, fmt.Errorf("invalid upload path") + } + + session := OcisSession{ + store: store, + info: tusd.FileInfo{}, + } + data, err := os.ReadFile(sessionPath) + if err != nil { + if errors.Is(err, iofs.ErrNotExist) { + // Interpret os.ErrNotExist as 404 Not Found + err = tusd.ErrNotFound + } + return nil, err + } + if err := json.Unmarshal(data, &session.info); err != nil { + return nil, err + } + + stat, err := os.Stat(session.binPath()) + if err != nil { + if os.IsNotExist(err) { + // Interpret os.ErrNotExist as 404 Not Found + err = tusd.ErrNotFound + } + return nil, err + } + + session.info.Offset = stat.Size() + + return &session, nil +} + +// Session is the interface used by the Cleanup call +type Session interface { + ID() string + Node(ctx context.Context) (*node.Node, error) + Context(ctx context.Context) context.Context + Cleanup(cleanNode, cleanBin, cleanInfo bool) +} + +// Cleanup cleans upload metadata, binary data and processing status as necessary +func (store OcisStore) Cleanup(ctx context.Context, session Session, failure bool, keepUpload bool) { + ctx, span := tracer.Start(session.Context(ctx), "Cleanup") + defer span.End() + session.Cleanup(failure, !keepUpload, !keepUpload) + + // unset processing status + n, err := session.Node(ctx) + if err != nil { + appctx.GetLogger(ctx).Info().Str("session", session.ID()).Err(err).Msg("could not read node") + return + } + // FIXME: after cleanup the node might already be deleted ... + if n != nil { // node can be nil when there was an error before it was created (eg. checksum-mismatch) + if err := n.UnmarkProcessing(ctx, session.ID()); err != nil { + appctx.GetLogger(ctx).Info().Str("path", n.InternalPath()).Err(err).Msg("unmarking processing failed") + } + } +} + +// CreateNodeForUpload will create the target node for the Upload +// TODO move this to the node package as NodeFromUpload? +// should we in InitiateUpload create the node first? and then the upload? +func (store OcisStore) CreateNodeForUpload(session *OcisSession, initAttrs node.Attributes) (*node.Node, error) { + ctx, span := tracer.Start(session.Context(context.Background()), "CreateNodeForUpload") + defer span.End() + n := node.New( + session.SpaceID(), + session.NodeID(), + session.NodeParentID(), + session.Filename(), + session.Size(), + session.ID(), + provider.ResourceType_RESOURCE_TYPE_FILE, + nil, + store.lu, + ) + var err error + n.SpaceRoot, err = node.ReadNode(ctx, store.lu, session.SpaceID(), session.SpaceID(), false, nil, false) + if err != nil { + return nil, err + } + + // check lock + if err := n.CheckLock(ctx); err != nil { + return nil, err + } + + var f *lockedfile.File + if session.NodeExists() { + f, err = store.updateExistingNode(ctx, session, n, session.SpaceID(), uint64(session.Size())) + if f != nil { + appctx.GetLogger(ctx).Info().Str("lockfile", f.Name()).Interface("err", err).Msg("got lock file from updateExistingNode") + } + } else { + f, err = store.initNewNode(ctx, session, n, uint64(session.Size())) + if f != nil { + appctx.GetLogger(ctx).Info().Str("lockfile", f.Name()).Interface("err", err).Msg("got lock file from initNewNode") + } + } + defer func() { + if f == nil { + return + } + if err := f.Close(); err != nil { + appctx.GetLogger(ctx).Error().Err(err).Str("nodeid", n.ID).Str("parentid", n.ParentID).Msg("could not close lock") + } + }() + if err != nil { + return nil, err + } + + mtime := time.Now() + if !session.MTime().IsZero() { + // overwrite mtime if requested + mtime = session.MTime() + } + + // overwrite technical information + initAttrs.SetString(prefixes.MTimeAttr, mtime.UTC().Format(time.RFC3339Nano)) + initAttrs.SetInt64(prefixes.TypeAttr, int64(provider.ResourceType_RESOURCE_TYPE_FILE)) + initAttrs.SetString(prefixes.ParentidAttr, n.ParentID) + initAttrs.SetString(prefixes.NameAttr, n.Name) + initAttrs.SetString(prefixes.BlobIDAttr, n.BlobID) + initAttrs.SetInt64(prefixes.BlobsizeAttr, n.Blobsize) + initAttrs.SetString(prefixes.StatusPrefix, node.ProcessingStatus+session.ID()) + + // update node metadata with new blobid etc + err = n.SetXattrsWithContext(ctx, initAttrs, false) + if err != nil { + return nil, errors.Wrap(err, "Decomposedfs: could not write metadata") + } + + if err := session.Persist(ctx); err != nil { + return nil, err + } + + return n, nil +} + +func (store OcisStore) initNewNode(ctx context.Context, session *OcisSession, n *node.Node, fsize uint64) (*lockedfile.File, error) { + // create folder structure (if needed) + if err := os.MkdirAll(filepath.Dir(n.InternalPath()), 0700); err != nil { + return nil, err + } + + // create and write lock new node metadata + f, err := lockedfile.OpenFile(store.lu.MetadataBackend().LockfilePath(n.InternalPath()), os.O_RDWR|os.O_CREATE, 0600) + if err != nil { + return nil, err + } + + // we also need to touch the actual node file here it stores the mtime of the resource + h, err := os.OpenFile(n.InternalPath(), os.O_CREATE|os.O_EXCL, 0600) + if err != nil { + return f, err + } + h.Close() + + if _, err := node.CheckQuota(ctx, n.SpaceRoot, false, 0, fsize); err != nil { + return f, err + } + + // link child name to parent if it is new + childNameLink := filepath.Join(n.ParentPath(), n.Name) + relativeNodePath := filepath.Join("../../../../../", lookup.Pathify(n.ID, 4, 2)) + log := appctx.GetLogger(ctx).With().Str("childNameLink", childNameLink).Str("relativeNodePath", relativeNodePath).Logger() + log.Info().Msg("initNewNode: creating symlink") + + if err = os.Symlink(relativeNodePath, childNameLink); err != nil { + log.Info().Err(err).Msg("initNewNode: symlink failed") + if errors.Is(err, iofs.ErrExist) { + log.Info().Err(err).Msg("initNewNode: symlink already exists") + return f, errtypes.AlreadyExists(n.Name) + } + return f, errors.Wrap(err, "Decomposedfs: could not symlink child entry") + } + log.Info().Msg("initNewNode: symlink created") + + // on a new file the sizeDiff is the fileSize + session.info.MetaData["sizeDiff"] = strconv.FormatInt(int64(fsize), 10) + return f, nil +} + +func (store OcisStore) updateExistingNode(ctx context.Context, session *OcisSession, n *node.Node, spaceID string, fsize uint64) (*lockedfile.File, error) { + targetPath := n.InternalPath() + + // write lock existing node before reading any metadata + f, err := lockedfile.OpenFile(store.lu.MetadataBackend().LockfilePath(targetPath), os.O_RDWR|os.O_CREATE, 0600) + if err != nil { + return nil, err + } + + old, _ := node.ReadNode(ctx, store.lu, spaceID, n.ID, false, nil, false) + if _, err := node.CheckQuota(ctx, n.SpaceRoot, true, uint64(old.Blobsize), fsize); err != nil { + return f, err + } + + oldNodeMtime, err := old.GetMTime(ctx) + if err != nil { + return f, err + } + oldNodeEtag, err := node.CalculateEtag(old.ID, oldNodeMtime) + if err != nil { + return f, err + } + + // When the if-match header was set we need to check if the + // etag still matches before finishing the upload. + if session.HeaderIfMatch() != "" && session.HeaderIfMatch() != oldNodeEtag { + return f, errtypes.Aborted("etag mismatch") + } + + // When the if-none-match header was set we need to check if any of the + // etags matches before finishing the upload. + if session.HeaderIfNoneMatch() != "" { + if session.HeaderIfNoneMatch() == "*" { + return f, errtypes.Aborted("etag mismatch, resource exists") + } + for _, ifNoneMatchTag := range strings.Split(session.HeaderIfNoneMatch(), ",") { + if ifNoneMatchTag == oldNodeEtag { + return f, errtypes.Aborted("etag mismatch") + } + } + } + + // When the if-unmodified-since header was set we need to check if the + // etag still matches before finishing the upload. + if session.HeaderIfUnmodifiedSince() != "" { + ifUnmodifiedSince, err := time.Parse(time.RFC3339Nano, session.HeaderIfUnmodifiedSince()) + if err != nil { + return f, errtypes.InternalError(fmt.Sprintf("failed to parse if-unmodified-since time: %s", err)) + } + + if oldNodeMtime.After(ifUnmodifiedSince) { + return f, errtypes.Aborted("if-unmodified-since mismatch") + } + } + + session.info.MetaData["versionsPath"] = session.store.lu.InternalPath(spaceID, n.ID+node.RevisionIDDelimiter+oldNodeMtime.UTC().Format(time.RFC3339Nano)) + session.info.MetaData["sizeDiff"] = strconv.FormatInt((int64(fsize) - old.Blobsize), 10) + + // create version node + if _, err := os.Create(session.info.MetaData["versionsPath"]); err != nil { + return f, err + } + + // copy blob metadata to version node + if err := store.lu.CopyMetadataWithSourceLock(ctx, targetPath, session.info.MetaData["versionsPath"], func(attributeName string, value []byte) (newValue []byte, copy bool) { + return value, strings.HasPrefix(attributeName, prefixes.ChecksumPrefix) || + attributeName == prefixes.TypeAttr || + attributeName == prefixes.BlobIDAttr || + attributeName == prefixes.BlobsizeAttr || + attributeName == prefixes.MTimeAttr + }, f, true); err != nil { + return f, err + } + + // keep mtime from previous version + if err := os.Chtimes(session.info.MetaData["versionsPath"], oldNodeMtime, oldNodeMtime); err != nil { + return f, errtypes.InternalError(fmt.Sprintf("failed to change mtime of version node: %s", err)) + } + + return f, nil +} diff --git a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/upload.go b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/upload.go index cb00376a5..53551696f 100644 --- a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/upload.go +++ b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/upload.go @@ -23,14 +23,12 @@ import ( "crypto/md5" "crypto/sha1" "encoding/hex" - "encoding/json" "fmt" "hash" "hash/adler32" "io" "io/fs" "os" - "path/filepath" "strings" "time" @@ -40,14 +38,10 @@ import ( "github.com/cs3org/reva/v2/pkg/errtypes" "github.com/cs3org/reva/v2/pkg/events" "github.com/cs3org/reva/v2/pkg/rhttp/datatx/metrics" - "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/lookup" "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata/prefixes" "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node" - "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/options" - "github.com/cs3org/reva/v2/pkg/utils" "github.com/golang-jwt/jwt" "github.com/pkg/errors" - "github.com/rs/zerolog" tusd "github.com/tus/tusd/pkg/handler" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/trace" @@ -80,78 +74,14 @@ type Tree interface { Propagate(ctx context.Context, node *node.Node, sizeDiff int64) (err error) } -// Upload processes the upload -// it implements tus tusd.Upload interface https://tus.io/protocols/resumable-upload.html#core-protocol -// it also implements its termination extension as specified in https://tus.io/protocols/resumable-upload.html#termination -// it also implements its creation-defer-length extension as specified in https://tus.io/protocols/resumable-upload.html#creation -// it also implements its concatenation extension as specified in https://tus.io/protocols/resumable-upload.html#concatenation -type Upload struct { - // we use a struct field on the upload as tus pkg will give us an empty context.Background - Ctx context.Context - // info stores the current information about the upload - Info tusd.FileInfo - // node for easy access - Node *node.Node - // SizeDiff size difference between new and old file version - SizeDiff int64 - // infoPath is the path to the .info file - infoPath string - // binPath is the path to the binary file (which has no extension) - binPath string - // lu and tp needed for file operations - lu *lookup.Lookup - tp Tree - // versionsPath will be empty if there was no file before - versionsPath string - // and a logger as well - log zerolog.Logger - // publisher used to publish events - pub events.Publisher - // async determines if uploads shoud be done asynchronously - async bool - // tknopts hold token signing information - tknopts options.TokenOptions -} - -func buildUpload(ctx context.Context, info tusd.FileInfo, binPath string, infoPath string, lu *lookup.Lookup, tp Tree, pub events.Publisher, async bool, tknopts options.TokenOptions) *Upload { - return &Upload{ - Info: info, - binPath: binPath, - infoPath: infoPath, - lu: lu, - tp: tp, - Ctx: ctx, - pub: pub, - async: async, - tknopts: tknopts, - log: appctx.GetLogger(ctx). - With(). - Interface("info", info). - Str("binPath", binPath). - Logger(), - } -} - -// Cleanup cleans the upload -func Cleanup(upload *Upload, failure bool, keepUpload bool) { - ctx, span := tracer.Start(upload.Ctx, "Cleanup") - defer span.End() - upload.cleanup(failure, !keepUpload, !keepUpload) - - // unset processing status - if upload.Node != nil { // node can be nil when there was an error before it was created (eg. checksum-mismatch) - if err := upload.Node.UnmarkProcessing(ctx, upload.Info.ID); err != nil { - upload.log.Info().Str("path", upload.Node.InternalPath()).Err(err).Msg("unmarking processing failed") - } - } -} +var defaultFilePerm = os.FileMode(0664) // WriteChunk writes the stream from the reader to the given offset of the upload -func (upload *Upload) WriteChunk(_ context.Context, offset int64, src io.Reader) (int64, error) { - ctx, span := tracer.Start(upload.Ctx, "WriteChunk") +func (session *OcisSession) WriteChunk(ctx context.Context, offset int64, src io.Reader) (int64, error) { + ctx, span := tracer.Start(session.Context(ctx), "WriteChunk") defer span.End() _, subspan := tracer.Start(ctx, "os.OpenFile") - file, err := os.OpenFile(upload.binPath, os.O_WRONLY|os.O_APPEND, defaultFilePerm) + file, err := os.OpenFile(session.binPath(), os.O_WRONLY|os.O_APPEND, defaultFilePerm) subspan.End() if err != nil { return 0, err @@ -174,32 +104,30 @@ func (upload *Upload) WriteChunk(_ context.Context, offset int64, src io.Reader) return n, err } - upload.Info.Offset += n - return n, upload.writeInfo() + // update upload.Session.Offset so subsequent code flow can use it. + // No need to persist the session as the offset is determined by stating the blob in the GetUpload / ReadSession codepath. + // The session offset is written to disk in FinishUpload + session.info.Offset += n + return n, nil } // GetInfo returns the FileInfo -func (upload *Upload) GetInfo(_ context.Context) (tusd.FileInfo, error) { - return upload.Info, nil +func (session *OcisSession) GetInfo(_ context.Context) (tusd.FileInfo, error) { + return session.ToFileInfo(), nil } // GetReader returns an io.Reader for the upload -func (upload *Upload) GetReader(_ context.Context) (io.Reader, error) { - _, span := tracer.Start(upload.Ctx, "GetReader") +func (session *OcisSession) GetReader(ctx context.Context) (io.Reader, error) { + _, span := tracer.Start(session.Context(ctx), "GetReader") defer span.End() - return os.Open(upload.binPath) + return os.Open(session.binPath()) } // FinishUpload finishes an upload and moves the file to the internal destination -func (upload *Upload) FinishUpload(_ context.Context) error { - ctx, span := tracer.Start(upload.Ctx, "FinishUpload") +func (session *OcisSession) FinishUpload(ctx context.Context) error { + ctx, span := tracer.Start(session.Context(ctx), "FinishUpload") defer span.End() - // set lockID to context - if upload.Info.MetaData["lockid"] != "" { - upload.Ctx = ctxpkg.ContextSetLockID(upload.Ctx, upload.Info.MetaData["lockid"]) - } - - log := appctx.GetLogger(upload.Ctx) + log := appctx.GetLogger(ctx) // calculate the checksum of the written bytes // they will all be written to the metadata later, so we cannot omit any of them @@ -210,11 +138,11 @@ func (upload *Upload) FinishUpload(_ context.Context) error { adler32h := adler32.New() { _, subspan := tracer.Start(ctx, "os.Open") - f, err := os.Open(upload.binPath) + f, err := os.Open(session.binPath()) subspan.End() if err != nil { // we can continue if no oc checksum header is set - log.Info().Err(err).Str("binPath", upload.binPath).Msg("error opening binPath") + log.Info().Err(err).Str("binPath", session.binPath()).Msg("error opening binPath") } defer f.Close() @@ -231,24 +159,25 @@ func (upload *Upload) FinishUpload(_ context.Context) error { // compare if they match the sent checksum // TODO the tus checksum extension would do this on every chunk, but I currently don't see an easy way to pass in the requested checksum. for now we do it in FinishUpload which is also called for chunked uploads - if upload.Info.MetaData["checksum"] != "" { + var err error + if session.info.MetaData["checksum"] != "" { var err error - parts := strings.SplitN(upload.Info.MetaData["checksum"], " ", 2) + parts := strings.SplitN(session.info.MetaData["checksum"], " ", 2) if len(parts) != 2 { return errtypes.BadRequest("invalid checksum format. must be '[algorithm] [checksum]'") } switch parts[0] { case "sha1": - err = upload.checkHash(parts[1], sha1h) + err = checkHash(parts[1], sha1h) case "md5": - err = upload.checkHash(parts[1], md5h) + err = checkHash(parts[1], md5h) case "adler32": - err = upload.checkHash(parts[1], adler32h) + err = checkHash(parts[1], adler32h) default: err = errtypes.BadRequest("unsupported checksum algorithm: " + parts[0]) } if err != nil { - Cleanup(upload, true, false) + session.store.Cleanup(ctx, session, true, false) return err } } @@ -260,9 +189,9 @@ func (upload *Upload) FinishUpload(_ context.Context) error { prefixes.ChecksumPrefix + "adler32": adler32h.Sum(nil), } - n, err := CreateNodeForUpload(upload, attrs) + n, err := session.store.CreateNodeForUpload(session, attrs) if err != nil { - Cleanup(upload, true, false) + session.store.Cleanup(ctx, session, true, false) return err } @@ -271,32 +200,30 @@ func (upload *Upload) FinishUpload(_ context.Context) error { metrics.UploadProcessing.Inc() metrics.UploadSessionsBytesReceived.Inc() - upload.Node = n - - if upload.pub != nil { - u, _ := ctxpkg.ContextGetUser(upload.Ctx) - s, err := upload.URL(upload.Ctx) + if session.store.pub != nil { + u, _ := ctxpkg.ContextGetUser(ctx) + s, err := session.URL(ctx) if err != nil { return err } - if err := events.Publish(ctx, upload.pub, events.BytesReceived{ - UploadID: upload.Info.ID, + if err := events.Publish(ctx, session.store.pub, events.BytesReceived{ + UploadID: session.ID(), URL: s, - SpaceOwner: n.SpaceOwnerOrManager(upload.Ctx), + SpaceOwner: n.SpaceOwnerOrManager(session.Context(ctx)), ExecutingUser: u, ResourceID: &provider.ResourceId{SpaceId: n.SpaceID, OpaqueId: n.ID}, - Filename: upload.Info.Storage["NodeName"], - Filesize: uint64(upload.Info.Size), + Filename: session.Filename(), + Filesize: uint64(session.Size()), }); err != nil { return err } } - if !upload.async { + if !session.store.async { // handle postprocessing synchronously - err = upload.Finalize() - Cleanup(upload, err != nil, false) + err = session.Finalize() + session.store.Cleanup(ctx, session, err != nil, false) if err != nil { log.Error().Err(err).Msg("failed to upload") return err @@ -304,34 +231,34 @@ func (upload *Upload) FinishUpload(_ context.Context) error { metrics.UploadSessionsFinalized.Inc() } - return upload.tp.Propagate(upload.Ctx, n, upload.SizeDiff) + return session.store.tp.Propagate(ctx, n, session.SizeDiff()) } // Terminate terminates the upload -func (upload *Upload) Terminate(_ context.Context) error { - upload.cleanup(true, true, true) +func (session *OcisSession) Terminate(_ context.Context) error { + session.Cleanup(true, true, true) return nil } // DeclareLength updates the upload length information -func (upload *Upload) DeclareLength(_ context.Context, length int64) error { - upload.Info.Size = length - upload.Info.SizeIsDeferred = false - return upload.writeInfo() +func (session *OcisSession) DeclareLength(ctx context.Context, length int64) error { + session.info.Size = length + session.info.SizeIsDeferred = false + return session.Persist(session.Context(ctx)) } // ConcatUploads concatenates multiple uploads -func (upload *Upload) ConcatUploads(_ context.Context, uploads []tusd.Upload) (err error) { - file, err := os.OpenFile(upload.binPath, os.O_WRONLY|os.O_APPEND, defaultFilePerm) +func (session *OcisSession) ConcatUploads(_ context.Context, uploads []tusd.Upload) (err error) { + file, err := os.OpenFile(session.binPath(), os.O_WRONLY|os.O_APPEND, defaultFilePerm) if err != nil { return err } defer file.Close() for _, partialUpload := range uploads { - fileUpload := partialUpload.(*Upload) + fileUpload := partialUpload.(*OcisSession) - src, err := os.Open(fileUpload.binPath) + src, err := os.Open(fileUpload.binPath()) if err != nil { return err } @@ -345,34 +272,18 @@ func (upload *Upload) ConcatUploads(_ context.Context, uploads []tusd.Upload) (e return } -// writeInfo updates the entire information. Everything will be overwritten. -func (upload *Upload) writeInfo() error { - _, span := tracer.Start(upload.Ctx, "writeInfo") +// Finalize finalizes the upload (eg moves the file to the internal destination) +func (session *OcisSession) Finalize() (err error) { + ctx, span := tracer.Start(session.Context(context.Background()), "Finalize") defer span.End() - data, err := json.Marshal(upload.Info) + n, err := session.Node(ctx) if err != nil { return err } - return os.WriteFile(upload.infoPath, data, defaultFilePerm) -} - -// Finalize finalizes the upload (eg moves the file to the internal destination) -func (upload *Upload) Finalize() (err error) { - ctx, span := tracer.Start(upload.Ctx, "Finalize") - defer span.End() - n := upload.Node - if n == nil { - var err error - n, err = node.ReadNode(ctx, upload.lu, upload.Info.Storage["SpaceRoot"], upload.Info.Storage["NodeId"], false, nil, false) - if err != nil { - return err - } - upload.Node = n - } // upload the data to the blobstore _, subspan := tracer.Start(ctx, "WriteBlob") - err = upload.tp.WriteBlob(n, upload.binPath) + err = session.store.tp.WriteBlob(n, session.binPath()) subspan.End() if err != nil { return errors.Wrap(err, "failed to upload file to blobstore") @@ -381,72 +292,77 @@ func (upload *Upload) Finalize() (err error) { return nil } -func (upload *Upload) checkHash(expected string, h hash.Hash) error { - if expected != hex.EncodeToString(h.Sum(nil)) { - return errtypes.ChecksumMismatch(fmt.Sprintf("invalid checksum: expected %s got %x", upload.Info.MetaData["checksum"], h.Sum(nil))) +func checkHash(expected string, h hash.Hash) error { + hash := hex.EncodeToString(h.Sum(nil)) + if expected != hash { + return errtypes.ChecksumMismatch(fmt.Sprintf("invalid checksum: expected %s got %x", expected, hash)) } return nil } +func (session *OcisSession) removeNode(ctx context.Context) { + n, err := session.Node(ctx) + if err != nil { + appctx.GetLogger(ctx).Error().Str("session", session.ID()).Err(err).Msg("getting node from session failed") + return + } + if err := n.Purge(ctx); err != nil { + appctx.GetLogger(ctx).Error().Str("nodepath", n.InternalPath()).Err(err).Msg("purging node failed") + } +} + // cleanup cleans up after the upload is finished -func (upload *Upload) cleanup(cleanNode, cleanBin, cleanInfo bool) { - if cleanNode && upload.Node != nil { - switch p := upload.versionsPath; p { - case "": - // remove node - if err := utils.RemoveItem(upload.Node.InternalPath()); err != nil { - upload.log.Info().Str("path", upload.Node.InternalPath()).Err(err).Msg("removing node failed") +func (session *OcisSession) Cleanup(cleanNode, cleanBin, cleanInfo bool) { + ctx := session.Context(context.Background()) + + if cleanNode { + if session.NodeExists() { + p := session.info.MetaData["versionsPath"] + n, err := session.Node(ctx) + if err != nil { + appctx.GetLogger(ctx).Error().Err(err).Str("sessionid", session.ID()).Msg("reading node for session failed") } - - // no old version was present - remove child entry - src := filepath.Join(upload.Node.ParentPath(), upload.Node.Name) - if err := os.Remove(src); err != nil { - upload.log.Info().Str("path", upload.Node.ParentPath()).Err(err).Msg("removing node from parent failed") - } - - // remove node from upload as it no longer exists - upload.Node = nil - default: - - if err := upload.lu.CopyMetadata(upload.Ctx, p, upload.Node.InternalPath(), func(attributeName string, value []byte) (newValue []byte, copy bool) { + if err := session.store.lu.CopyMetadata(ctx, p, n.InternalPath(), func(attributeName string, value []byte) (newValue []byte, copy bool) { return value, strings.HasPrefix(attributeName, prefixes.ChecksumPrefix) || attributeName == prefixes.TypeAttr || attributeName == prefixes.BlobIDAttr || attributeName == prefixes.BlobsizeAttr || attributeName == prefixes.MTimeAttr }, true); err != nil { - upload.log.Info().Str("versionpath", p).Str("nodepath", upload.Node.InternalPath()).Err(err).Msg("renaming version node failed") + appctx.GetLogger(ctx).Info().Str("versionpath", p).Str("nodepath", n.InternalPath()).Err(err).Msg("renaming version node failed") } if err := os.RemoveAll(p); err != nil { - upload.log.Info().Str("versionpath", p).Str("nodepath", upload.Node.InternalPath()).Err(err).Msg("error removing version") + appctx.GetLogger(ctx).Info().Str("versionpath", p).Str("nodepath", n.InternalPath()).Err(err).Msg("error removing version") } + } else { + session.removeNode(ctx) } } if cleanBin { - if err := os.Remove(upload.binPath); err != nil && !errors.Is(err, fs.ErrNotExist) { - upload.log.Error().Str("path", upload.binPath).Err(err).Msg("removing upload failed") + if err := os.Remove(session.binPath()); err != nil && !errors.Is(err, fs.ErrNotExist) { + appctx.GetLogger(ctx).Error().Str("path", session.binPath()).Err(err).Msg("removing upload failed") } } if cleanInfo { - if err := os.Remove(upload.infoPath); err != nil && !errors.Is(err, fs.ErrNotExist) { - upload.log.Error().Str("path", upload.infoPath).Err(err).Msg("removing upload info failed") + if err := session.Purge(ctx); err != nil && !errors.Is(err, fs.ErrNotExist) { + appctx.GetLogger(ctx).Error().Err(err).Str("session", session.ID()).Msg("removing upload info failed") } } } // URL returns a url to download an upload -func (upload *Upload) URL(_ context.Context) (string, error) { +func (session *OcisSession) URL(_ context.Context) (string, error) { type transferClaims struct { jwt.StandardClaims Target string `json:"target"` } - u := joinurl(upload.tknopts.DownloadEndpoint, "tus/", upload.Info.ID) - ttl := time.Duration(upload.tknopts.TransferExpires) * time.Second + u := joinurl(session.store.tknopts.DownloadEndpoint, "tus/", session.ID()) + ttl := time.Duration(session.store.tknopts.TransferExpires) * time.Second claims := transferClaims{ StandardClaims: jwt.StandardClaims{ ExpiresAt: time.Now().Add(ttl).Unix(), @@ -458,12 +374,12 @@ func (upload *Upload) URL(_ context.Context) (string, error) { t := jwt.NewWithClaims(jwt.GetSigningMethod("HS256"), claims) - tkn, err := t.SignedString([]byte(upload.tknopts.TransferSharedSecret)) + tkn, err := t.SignedString([]byte(session.store.tknopts.TransferSharedSecret)) if err != nil { return "", errors.Wrapf(err, "error signing token with claims %+v", claims) } - return joinurl(upload.tknopts.DataGatewayEndpoint, tkn), nil + return joinurl(session.store.tknopts.DataGatewayEndpoint, tkn), nil } // replace with url.JoinPath after switching to go1.19 diff --git a/vendor/modules.txt b/vendor/modules.txt index afbeadac3..c572f7a92 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -362,7 +362,7 @@ github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1 github.com/cs3org/go-cs3apis/cs3/storage/registry/v1beta1 github.com/cs3org/go-cs3apis/cs3/tx/v1beta1 github.com/cs3org/go-cs3apis/cs3/types/v1beta1 -# github.com/cs3org/reva/v2 v2.17.1-0.20231220115644-93b4dd91a8b3 +# github.com/cs3org/reva/v2 v2.17.1-0.20231222094355-457e00743c93 ## explicit; go 1.21 github.com/cs3org/reva/v2/cmd/revad/internal/grace github.com/cs3org/reva/v2/cmd/revad/runtime