operations: check checksums in rcat with known size - fixes #6305
RcatSize streams the body straight into Put behind an ObjectInfo with no hashes, so a known-size upload had its size checked but its checksum never verified, unlike the unknown-size path through Rcat which hashes as it goes. The stream can only be read once, so hash it on the way past and compare with the destination after the upload, as operations.Copy does. A destination which reports no usable hash is still checked by size, and one which fails either check has the failed copy removed.
This commit is contained in:
phatlc
Nick Craig-Wood
@@ -1818,8 +1818,20 @@ func RcatSize(ctx context.Context, fdst fs.Fs, dstFileName string, in io.ReadClo
|
||||
defer func() {
|
||||
tr.Done(ctx, err)
|
||||
}()
|
||||
body := io.NopCloser(in) // we let the server close the body
|
||||
in := tr.Account(ctx, body) // account the transfer (no buffering)
|
||||
// The stream can only be read once, so hash it on the way past
|
||||
// and compare with the destination afterwards.
|
||||
hashType, hashOption := CommonHash(ctx, fdst, fdst)
|
||||
var hasher *hash.MultiHasher
|
||||
var streamIn io.Reader = in
|
||||
if hashType != hash.None {
|
||||
hasher, err = hash.NewMultiHasherTypes(hashOption.Hashes)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
streamIn = io.TeeReader(streamIn, hasher)
|
||||
}
|
||||
body := io.NopCloser(streamIn) // we let the server close the body
|
||||
in := tr.Account(ctx, body) // account the transfer (no buffering)
|
||||
|
||||
if SkipDestructive(ctx, dstFileName, "upload from pipe") {
|
||||
// prevents "broken pipe" errors
|
||||
@@ -1827,7 +1839,7 @@ func RcatSize(ctx context.Context, fdst fs.Fs, dstFileName string, in io.ReadClo
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var options []fs.OpenOption
|
||||
options := []fs.OpenOption{hashOption}
|
||||
for _, option := range fs.GetConfig(ctx).UploadHeaders {
|
||||
options = append(options, option)
|
||||
}
|
||||
@@ -1839,13 +1851,25 @@ func RcatSize(ctx context.Context, fdst fs.Fs, dstFileName string, in io.ReadClo
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Check transfer - the source may have ended before size
|
||||
// bytes in which case the object will have been truncated
|
||||
// Verify the upload
|
||||
if sizeDiffers(ctx, info, obj) {
|
||||
err = fmt.Errorf("corrupted on transfer: sizes differ src %d vs dst(%s) %d", info.Size(), fdst, obj.Size())
|
||||
err = fmt.Errorf("corrupted on transfer: sizes differ src %d vs dst(%s) %d", size, obj.Fs(), obj.Size())
|
||||
} else if hasher != nil {
|
||||
src := object.NewStaticObjectInfo(dstFileName, modTime, size, true, hasher.Sums(), fdst)
|
||||
// checkHashes logs and counts errors
|
||||
same, _, srcHash, dstHash, _ := checkHashes(ctx, src, obj, hashType)
|
||||
if !same {
|
||||
err = fmt.Errorf("corrupted on transfer: %v hashes differ src %q vs dst(%s) %q", hashType, srcHash, obj.Fs(), dstHash)
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
err = fs.CountError(ctx, err)
|
||||
fs.Errorf(obj, "%v", err)
|
||||
return obj, err
|
||||
fs.Infof(obj, "Removing failed copy")
|
||||
if removeErr := obj.Remove(ctx); removeErr != nil {
|
||||
fs.Infof(obj, "Failed to remove failed copy: %s", removeErr)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
} else {
|
||||
// Size unknown use Rcat
|
||||
|
||||
@@ -39,6 +39,7 @@ import (
|
||||
"github.com/rclone/rclone/fs/filter"
|
||||
"github.com/rclone/rclone/fs/fshttp"
|
||||
"github.com/rclone/rclone/fs/hash"
|
||||
"github.com/rclone/rclone/fs/object"
|
||||
"github.com/rclone/rclone/fs/operations"
|
||||
"github.com/rclone/rclone/fstest"
|
||||
"github.com/rclone/rclone/fstest/fstests"
|
||||
@@ -1739,6 +1740,97 @@ func TestRcatSizeUploadHeaders(t *testing.T) {
|
||||
assert.True(t, found, "X-Upload-Header not found in options passed to Put")
|
||||
}
|
||||
|
||||
// corruptingFs wraps an fs.Fs, storing data which doesn't match the
|
||||
// data streamed into Put.
|
||||
type corruptingFs struct {
|
||||
fs.Fs
|
||||
}
|
||||
|
||||
func (f *corruptingFs) Put(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (fs.Object, error) {
|
||||
n, err := io.Copy(io.Discard, in)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return f.Fs.Put(ctx, bytes.NewReader(bytes.Repeat([]byte("!"), int(n))), src, options...)
|
||||
}
|
||||
|
||||
// noHashFs wraps an fs.Fs, reporting that it supports no hashes.
|
||||
type noHashFs struct {
|
||||
fs.Fs
|
||||
}
|
||||
|
||||
func (f *noHashFs) Hashes() hash.Set {
|
||||
return hash.NewHashSet()
|
||||
}
|
||||
|
||||
// truncatingFs wraps an fs.Fs, storing one byte less than was streamed
|
||||
// into Put and reporting that it supports no hashes, so only the size
|
||||
// can reveal the corruption.
|
||||
type truncatingFs struct {
|
||||
fs.Fs
|
||||
}
|
||||
|
||||
func (f *truncatingFs) Hashes() hash.Set {
|
||||
return hash.NewHashSet()
|
||||
}
|
||||
|
||||
func (f *truncatingFs) Put(ctx context.Context, in io.Reader, src fs.ObjectInfo, options ...fs.OpenOption) (fs.Object, error) {
|
||||
n, err := io.Copy(io.Discard, in)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
short := bytes.Repeat([]byte("!"), int(n)-1)
|
||||
info := object.NewStaticObjectInfo(src.Remote(), src.ModTime(ctx), int64(len(short)), true, nil, f.Fs)
|
||||
return f.Fs.Put(ctx, bytes.NewReader(short), info, options...)
|
||||
}
|
||||
|
||||
func TestRcatSizeChecksum(t *testing.T) {
|
||||
const body = "------------------------------------------------------------"
|
||||
|
||||
t.Run("Corrupted", func(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
r := fstest.NewRun(t)
|
||||
if r.Fremote.Hashes().Count() == 0 {
|
||||
t.Skip("Skipping as destination doesn't support hashes")
|
||||
}
|
||||
bodyReader := io.NopCloser(strings.NewReader(body))
|
||||
_, err := operations.RcatSize(ctx, &corruptingFs{Fs: r.Fremote}, "potato1", bodyReader, int64(len(body)), t1, nil)
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "corrupted on transfer")
|
||||
r.CheckRemoteItems(t)
|
||||
})
|
||||
|
||||
// A destination which reports no hashes can only be checked by its
|
||||
// size, so the size has to be compared as well as the hash.
|
||||
t.Run("SizeDiffers", func(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
r := fstest.NewRun(t)
|
||||
bodyReader := io.NopCloser(strings.NewReader(body))
|
||||
_, err := operations.RcatSize(ctx, &truncatingFs{Fs: r.Fremote}, "potato4", bodyReader, int64(len(body)), t1, nil)
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "corrupted on transfer: sizes differ")
|
||||
r.CheckRemoteItems(t)
|
||||
})
|
||||
|
||||
t.Run("IgnoreChecksum", func(t *testing.T) {
|
||||
ctx, ci := fs.AddConfig(context.Background())
|
||||
ci.IgnoreChecksum = true
|
||||
r := fstest.NewRun(t)
|
||||
bodyReader := io.NopCloser(strings.NewReader(body))
|
||||
_, err := operations.RcatSize(ctx, &corruptingFs{Fs: r.Fremote}, "potato2", bodyReader, int64(len(body)), t1, nil)
|
||||
require.NoError(t, err)
|
||||
})
|
||||
|
||||
t.Run("NoHashes", func(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
r := fstest.NewRun(t)
|
||||
bodyReader := io.NopCloser(strings.NewReader(body))
|
||||
obj, err := operations.RcatSize(ctx, &noHashFs{Fs: r.Fremote}, "potato3", bodyReader, int64(len(body)), t1, nil)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, int64(len(body)), obj.Size())
|
||||
})
|
||||
}
|
||||
|
||||
func TestTouchDir(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
r := fstest.NewRun(t)
|
||||
|
||||
Reference in New Issue
Block a user