sevenhill/QSfera
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bump reva (#8058)

Browse Source 
Signed-off-by: jkoberg <jkoberg@owncloud.com>
This commit is contained in:
kobergj
2023-12-22 15:10:39 +01:00
committed by GitHub
parent dee53ce6e4
commit 5362f82efa
18 changed files with 1130 additions and 947 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changelog/unreleased/bump-reva.md
+1
View File
@@ -3,3 +3,4 @@ Enhancement: Bump reva
Bumps reva version
https://github.com/owncloud/ocis/pull/8038
https://github.com/owncloud/ocis/pull/8056
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
vendor/github.com/cs3org/reva/v2/internal/grpc/services/publicshareprovider/publicshareprovider.go
Generated Vendored
+5 -2
View File
@@ -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(),
vendor/github.com/cs3org/reva/v2/internal/grpc/services/sharesstorageprovider/sharesstorageprovider.go
Generated Vendored
+21 -1
View File
@@ -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) {
vendor/github.com/cs3org/reva/v2/internal/grpc/services/usershareprovider/usershareprovider.go
Generated Vendored
+62 -4
View File
@@ -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{
vendor/github.com/cs3org/reva/v2/pkg/rhttp/datatx/manager/tus/tus.go
Generated Vendored
+13 -11
View File
@@ -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")
}
}
}
}
vendor/github.com/cs3org/reva/v2/pkg/share/manager/jsoncs3/receivedsharecache/receivedsharecache.go
Generated Vendored
+23 -20
View File
@@ -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)
}
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/decomposedfs.go
Generated Vendored
+43 -23
View File
@@ -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+"'")
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node/node.go
Generated Vendored
+16 -9
View File
@@ -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)
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/revisions.go
Generated Vendored
+1 -1
View File
@@ -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")
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/spaces.go
Generated Vendored
+1 -1
View File
@@ -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
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload.go
Generated Vendored
+148 -112
View File
@@ -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)
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/processing.go
Generated Vendored
-582
View File
@@ -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)
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/session.go
Generated Vendored
+319
View File
@@ -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")
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/store.go
Generated Vendored
+380
View File
@@ -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
}
vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload/upload.go
Generated Vendored
+93 -177
View File
@@ -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
vendor/modules.txt
Vendored
+1 -1
View File
@@ -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