diff --git a/backend/linkbox/linkbox.go b/backend/linkbox/linkbox.go index a91eb8857..47856f74c 100644 --- a/backend/linkbox/linkbox.go +++ b/backend/linkbox/linkbox.go @@ -11,10 +11,10 @@ package linkbox */ import ( - "bytes" "context" "crypto/md5" "crypto/tls" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -35,7 +35,9 @@ import ( "github.com/rclone/rclone/fs/fshttp" "github.com/rclone/rclone/fs/hash" "github.com/rclone/rclone/lib/dircache" + "github.com/rclone/rclone/lib/multipart" "github.com/rclone/rclone/lib/pacer" + "github.com/rclone/rclone/lib/pool" "github.com/rclone/rclone/lib/rest" ) @@ -46,7 +48,8 @@ const ( pacerBurst = 1 linkboxAPIURL = "https://www.linkbox.to/api/open/" linkboxWebAPIURL = "https://www.linkbox.to/api/" - rootID = "0" // ID of root directory + rootID = "0" // ID of root directory + prefixSize = 10 * 1024 * 1024 // the API wants the MD5 of the first 10 MiB of a file ) func init() { @@ -667,6 +670,21 @@ func (o *Object) Open(ctx context.Context, options ...fs.OpenOption) (io.ReadClo return res.Body, nil } +// readPrefix reads up to prefixSize bytes of in into a buffer from the +// global pool returning it along with the hex MD5 of what was read. +// +// The caller must Close the returned buffer when done with it. +func readPrefix(in io.Reader) (rw *pool.RW, md5sum string, err error) { + rw = multipart.NewRW() + hasher := md5.New() + _, err = io.CopyN(rw, io.TeeReader(in, hasher), prefixSize) + if err != nil && err != io.EOF { + _ = rw.Close() + return nil, "", err + } + return rw, hex.EncodeToString(hasher.Sum(nil)), nil +} + // Update in to the object with the modTime given of the given size // // When called from outside an Fs by rclone, src.Size() will always be >= 0. @@ -701,11 +719,13 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op } } - first10m := io.LimitReader(in, 10_485_760) - first10mBytes, err := io.ReadAll(first10m) + prefix, prefixMD5, err := readPrefix(in) if err != nil { return fmt.Errorf("Update err in reading file: %w", err) } + defer func() { + _ = prefix.Close() + }() // get upload authorization (step 1) opts := &rest.Opts{ @@ -715,7 +735,7 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op Options: options, Parameters: url.Values{ "token": {o.fs.opt.Token}, - "fileMd5ofPre10m": {fmt.Sprintf("%x", md5.Sum(first10mBytes))}, + "fileMd5ofPre10m": {prefixMD5}, "fileSize": {itoa64(size)}, }, } @@ -757,7 +777,7 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op return fmt.Errorf("head upload URL: %w", err) } - file := io.MultiReader(bytes.NewReader(first10mBytes), in) + file := io.MultiReader(prefix, in) opts.Method = "PUT" opts.Body = file @@ -797,7 +817,7 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op Options: options, Parameters: url.Values{ "token": {o.fs.opt.Token}, - "fileMd5ofPre10m": {fmt.Sprintf("%x", md5.Sum(first10mBytes))}, + "fileMd5ofPre10m": {prefixMD5}, "fileSize": {itoa64(size)}, "pid": {dirID}, "diyName": {leaf}, diff --git a/backend/linkbox/linkbox_internal_test.go b/backend/linkbox/linkbox_internal_test.go new file mode 100644 index 000000000..588ac6fd8 --- /dev/null +++ b/backend/linkbox/linkbox_internal_test.go @@ -0,0 +1,41 @@ +package linkbox + +import ( + "bytes" + "crypto/md5" + "fmt" + "io" + "testing" + + "github.com/rclone/rclone/lib/pool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestReadPrefix(t *testing.T) { + for _, size := range []int64{1, 1024, prefixSize - 1, prefixSize, prefixSize + 1, 3 * prefixSize} { + t.Run(fmt.Sprintf("%d", size), func(t *testing.T) { + inUse := pool.Global().InUse() + content := bytes.Repeat([]byte("x"), int(size)) + for i := range content { + content[i] = byte(i) + } + wantPrefix := content[:min(size, prefixSize)] + wantMD5 := fmt.Sprintf("%x", md5.Sum(wantPrefix)) + + in := bytes.NewReader(content) + rw, gotMD5, err := readPrefix(in) + require.NoError(t, err) + assert.Equal(t, wantMD5, gotMD5) + assert.Equal(t, int64(len(wantPrefix)), rw.Size()) + + // The prefix followed by the rest of in must reproduce the content + got, err := io.ReadAll(io.MultiReader(rw, in)) + require.NoError(t, err) + assert.Equal(t, content, got) + + require.NoError(t, rw.Close()) + assert.Equal(t, inUse, pool.Global().InUse(), "pool buffers leaked") + }) + } +}