diff --git a/block/block_formatter.go b/block/block_formatter.go index 80fc2ce52..d6896f6c3 100644 --- a/block/block_formatter.go +++ b/block/block_formatter.go @@ -17,10 +17,10 @@ type Formatter interface { // ComputeBlockID computes ID of the storage block for the specified block of data and returns it in ObjectID. ComputeBlockID(data []byte) []byte - // Encrypt returns encrypted bytes corresponding to the given plaintext. May reuse the input slice. + // Encrypt returns encrypted bytes corresponding to the given plaintext. Must not clobber the input slice. Encrypt(plainText []byte, blockID []byte) ([]byte, error) - // Decrypt returns unencrypted bytes corresponding to the given ciphertext. May reuse the input slice. + // Decrypt returns unencrypted bytes corresponding to the given ciphertext. Must not clobber the input slice. Decrypt(cipherText []byte, blockID []byte) ([]byte, error) } @@ -37,11 +37,11 @@ func (fi *unencryptedFormat) ComputeBlockID(data []byte) []byte { } func (fi *unencryptedFormat) Encrypt(plainText []byte, blockID []byte) ([]byte, error) { - return plainText, nil + return cloneBytes(plainText), nil } func (fi *unencryptedFormat) Decrypt(cipherText []byte, blockID []byte) ([]byte, error) { - return cipherText, nil + return cloneBytes(cipherText), nil } // syntheticIVEncryptionFormat implements encrypted format with single master AES key and StorageBlock==IV that's @@ -71,8 +71,9 @@ func symmetricEncrypt(createCipher func(key []byte) (cipher.Block, error), key [ } ctr := cipher.NewCTR(blockCipher, iv[0:blockCipher.BlockSize()]) - ctr.XORKeyStream(b, b) - return b, nil + result := make([]byte, len(b)) + ctr.XORKeyStream(result, b) + return result, nil } // SupportedFormats is a list of supported object formats including: diff --git a/block/block_manager.go b/block/block_manager.go index cc27b43fd..fd2c65765 100644 --- a/block/block_manager.go +++ b/block/block_manager.go @@ -444,7 +444,7 @@ func (bm *Manager) loadPackIndexesUnlocked(ctx context.Context) ([]IndexInfo, bo return nil, false, err } - err = bm.tryLoadPackIndexBlocksLocked(ctx, blocks) + err = bm.tryLoadPackIndexBlocksUnlocked(ctx, blocks) if err == nil { var blockIDs []string for _, b := range blocks { @@ -465,8 +465,8 @@ func (bm *Manager) loadPackIndexesUnlocked(ctx context.Context) ([]IndexInfo, bo return nil, false, fmt.Errorf("unable to load pack indexes despite %v retries", indexLoadAttempts) } -func (bm *Manager) tryLoadPackIndexBlocksLocked(ctx context.Context, blocks []IndexInfo) error { - ch, err := bm.unprocessedIndexBlocks(blocks) +func (bm *Manager) tryLoadPackIndexBlocksUnlocked(ctx context.Context, blocks []IndexInfo) error { + ch, err := bm.unprocessedIndexBlocksUnlocked(blocks) if err != nil { return err } @@ -509,8 +509,8 @@ func (bm *Manager) tryLoadPackIndexBlocksLocked(ctx context.Context, blocks []In return nil } -// unprocessedIndexBlocks returns a closed channel filled with block IDs that are not in committedBlocks cache. -func (bm *Manager) unprocessedIndexBlocks(blocks []IndexInfo) (<-chan string, error) { +// unprocessedIndexBlocksUnlocked returns a closed channel filled with block IDs that are not in committedBlocks cache. +func (bm *Manager) unprocessedIndexBlocksUnlocked(blocks []IndexInfo) (<-chan string, error) { ch := make(chan string, len(blocks)) for _, block := range blocks { has, err := bm.committedBlocks.cache.hasIndexBlockID(block.FileName) diff --git a/block/block_manager_test.go b/block/block_manager_test.go index 9af07e271..dcc28401e 100644 --- a/block/block_manager_test.go +++ b/block/block_manager_test.go @@ -5,6 +5,7 @@ import ( "context" "crypto/md5" "encoding/hex" + "errors" "fmt" "math/rand" "reflect" @@ -255,6 +256,48 @@ func TestBlockManagerWriteMultiple(t *testing.T) { } } +// This is regression test for a bug where we would corrupt data when encryption +// was done in place and clobbered pending data in memory. +func TestBlockManagerFailedToWritePack(t *testing.T) { + ctx := context.Background() + data := map[string][]byte{} + keyTime := map[string]time.Time{} + st := storagetesting.NewMapStorage(data, keyTime, nil) + faulty := &storagetesting.FaultyStorage{ + Base: st, + } + st = faulty + + bm, err := newManagerWithOptions(context.Background(), st, FormattingOptions{ + Version: 1, + BlockFormat: "ENCRYPTED_HMAC_SHA256_AES256_SIV", + MaxPackSize: maxPackSize, + HMACSecret: []byte("foo"), + MasterKey: []byte("0123456789abcdef0123456789abcdef"), + }, CachingOptions{}, fakeTimeNowFrozen(fakeTime)) + if err != nil { + t.Fatalf("can't create bm: %v", err) + } + logging.SetLevel(logging.DEBUG, "faulty-storage") + + faulty.Faults = map[string][]*storagetesting.Fault{ + "PutBlock": []*storagetesting.Fault{ + {Err: errors.New("booboo")}, + }, + } + + b1, err := bm.WriteBlock(ctx, seededRandomData(1, 10), "") + if err != nil { + t.Fatalf("can't create block: %v", err) + } + + if err := bm.Flush(ctx); err != nil { + t.Logf("expected flush error: %v", err) + } + + verifyBlock(ctx, t, bm, b1, seededRandomData(1, 10)) +} + func TestBlockManagerConcurrency(t *testing.T) { ctx := context.Background() data := map[string][]byte{} diff --git a/block/committed_block_index.go b/block/committed_block_index.go index 3b22b64ed..354fe138b 100644 --- a/block/committed_block_index.go +++ b/block/committed_block_index.go @@ -65,7 +65,7 @@ func (b *committedBlockIndex) addBlock(indexBlockID string, data []byte, use boo func (b *committedBlockIndex) listBlocks(prefix string, cb func(i Info) error) error { b.mu.Lock() - m := b.merged + m := append(packindex.Merged(nil), b.merged...) b.mu.Unlock() return m.Iterate(prefix, cb) diff --git a/internal/storagetesting/faulty.go b/internal/storagetesting/faulty.go new file mode 100644 index 000000000..8043c1488 --- /dev/null +++ b/internal/storagetesting/faulty.go @@ -0,0 +1,101 @@ +package storagetesting + +import ( + "context" + "sync" + "time" + + "github.com/kopia/kopia/internal/kopialogging" + "github.com/kopia/kopia/storage" +) + +var log = kopialogging.Logger("faulty-storage") + +type Fault struct { + Repeat int + Sleep time.Duration + WaitFor chan struct{} // waits until the given channel is closed before returning + Err error +} + +type FaultyStorage struct { + Base storage.Storage + Faults map[string][]*Fault + + mu sync.Mutex +} + +func (s *FaultyStorage) GetBlock(ctx context.Context, id string, offset, length int64) ([]byte, error) { + if err := s.getNextFault("GetBlock", id, offset, length); err != nil { + return nil, err + } + return s.Base.GetBlock(ctx, id, offset, length) +} + +func (s *FaultyStorage) PutBlock(ctx context.Context, id string, data []byte) error { + if err := s.getNextFault("PutBlock", id, len(data)); err != nil { + return err + } + return s.Base.PutBlock(ctx, id, data) +} + +func (s *FaultyStorage) DeleteBlock(ctx context.Context, id string) error { + if err := s.getNextFault("DeleteBlock", id); err != nil { + return err + } + return s.Base.DeleteBlock(ctx, id) +} + +func (s *FaultyStorage) ListBlocks(ctx context.Context, prefix string, callback func(storage.BlockMetadata) error) error { + if err := s.getNextFault("ListBlocks", prefix); err != nil { + return err + } + + return s.Base.ListBlocks(ctx, prefix, func(bm storage.BlockMetadata) error { + if err := s.getNextFault("ListBlocksItem", prefix); err != nil { + return err + } + return callback(bm) + }) +} + +func (s *FaultyStorage) Close(ctx context.Context) error { + if err := s.getNextFault("Close"); err != nil { + return err + } + return s.Base.Close(ctx) +} + +func (s *FaultyStorage) ConnectionInfo() storage.ConnectionInfo { + return s.Base.ConnectionInfo() +} + +func (s *FaultyStorage) getNextFault(method string, args ...interface{}) error { + s.mu.Lock() + faults := s.Faults[method] + if len(faults) == 0 { + s.mu.Unlock() + log.Debugf("no faults for %v %v", method, args) + return nil + } + + f := faults[0] + if f.Repeat > 0 { + f.Repeat-- + log.Debugf("will repeat %v more times the fault for %v %v", f.Repeat, method, args) + } else { + s.Faults[method] = faults[1:] + } + s.mu.Unlock() + if f.WaitFor != nil { + log.Debugf("waiting for channel to be closed in %v %v", method, args) + <-f.WaitFor + } + if f.Sleep > 0 { + log.Debugf("sleeping for %v in %v %v", f.Sleep, method, args) + } + log.Debugf("returning %v for %v %v", f.Err, method, args) + return f.Err +} + +var _ storage.Storage = (*FaultyStorage)(nil)