fix: incomplete streamReader read (#30)

Signed-off-by: francois  samin <[email protected]>

Author: fsamin

convergent/convergent_test.go

@@ -5,11 +5,16 @@ import (
 	"crypto/rand"
 	"encoding/hex"
 	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/ovh/configstore"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/ovh/symmecrypt/ciphers/aesgcm"
@@ -257,3 +262,59 @@ func TestLoadKeyFromStore(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, k)
 }
+
+func TestDecryptFromHTTP(t *testing.T) {
+	// Key config
+	cfgs := []convergent.ConvergentEncryptionConfig{
+		{
+			Cipher:      aesgcm.CipherName,
+			LocatorSalt: symutils.RandomSalt(),
+			SecretValue: symutils.MustRandomString(10),
+		},
+	}
+
+	// Encrypt a random content
+	clearContent := make([]byte, 10*1024)
+	rand.Read(clearContent) // nolint
+
+	h, err := convergent.NewHash(bytes.NewReader(clearContent))
+	require.NoError(t, err)
+
+	k, err := convergent.NewKey(h, cfgs...)
+	require.NoError(t, err)
+	require.NotNil(t, k)
+
+	dest := new(bytes.Buffer)
+	err = k.EncryptPipe(bytes.NewReader(clearContent), dest)
+	require.NoError(t, err)
+	encryptedContent := dest.String()
+
+	// Serve the encrypted content
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		time.Sleep(1 * time.Second)
+		io.WriteString(w, encryptedContent) //nolint
+	}))
+	defer ts.Close()
+
+	wg := &sync.WaitGroup{}
+
+	for i := 0; i < 100; i++ {
+		wg.Add(1)
+		go func() {
+			res, err := http.Get(ts.URL)
+			require.NoError(t, err)
+
+			defer res.Body.Close()
+
+			dest := new(bytes.Buffer)
+			err = k.DecryptPipe(res.Body, dest)
+			require.NoError(t, err)
+
+			// Ensure the content is correctly decrypted
+			assert.EqualValues(t, clearContent, dest.Bytes())
+			wg.Done()
+		}()
+	}
+
+	wg.Wait()
+}

stream/stream.go

@@ -177,23 +177,23 @@ func NewReader(r io.Reader, k symmecrypt.Key, chunkSize int, extras ...[]byte) i
 
 func (r *chunksReader) readNewChunk() error {
 	// read the chunksize
-	headerBtes := make([]byte, binary.MaxVarintLen32)
-	if _, err := r.src.Read(headerBtes); err != nil { // READING THE CLEAR HEADER FROM THE ENCRYPTED SOURCE
+	var headerBtes = new(bytes.Buffer)
+	if _, err := io.CopyN(headerBtes, r.src, binary.MaxVarintLen32); err != nil {
 		return err
 	}
 
-	n, err := binary.ReadUvarint(bytes.NewReader(headerBtes)) // READ THE HEADER BUFFER
+	n, err := binary.ReadUvarint(bytes.NewReader(headerBtes.Bytes())) // READ THE HEADER BUFFER
 	if err != nil {
 		return err
 	}
 
 	// read the chunk content
-	btes := make([]byte, n)
-	_, err = r.src.Read(btes)
-	if err != nil && err != io.EOF {
+	var btsBuff = new(bytes.Buffer)
+	if _, err := io.CopyN(btsBuff, r.src, int64(n)); err != nil && err != io.EOF {
 		return err
 	}
 
+	var btes = btsBuff.Bytes()
 	var clearContent []byte
 
 	if r.uncappedK == nil {
@@ -225,7 +225,7 @@ func (r *chunksReader) Read(p []byte) (x int, e error) {
 		}
 	}
 
-	if len(p)+r.currentChunkReadBytes > r.chunkSize {
+	if len(p)+r.currentChunkReadBytes >= r.chunkSize {
 		var pp = p
 		for {
 			// The first part of 'p' will store the current chunk

stream/stream_test.go

@@ -0,0 +1,74 @@
+package stream_test
+
+import (
+	"bytes"
+	"crypto/rand"
+	"io"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/ovh/configstore"
+	"github.com/ovh/symmecrypt/keyloader"
+	"github.com/ovh/symmecrypt/stream"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func ProviderTest() (configstore.ItemList, error) {
+	ret := configstore.ItemList{
+		Items: []configstore.Item{
+			configstore.NewItem(
+				keyloader.EncryptionKeyConfigName,
+				`{"key":"5fdb8af280b007a46553dfddb3f42bc10619dcabca8d4fdf5239b09445ab1a41","identifier":"test","sealed":false,"timestamp":1522325806,"cipher":"aes-gcm"}`,
+				1,
+			),
+			configstore.NewItem(
+				keyloader.EncryptionKeyConfigName,
+				`{"key":"7db2b4b695e11563edca94b0f9c7ad16919fc11eac414c1b1706cbaa3c3e61a4b884301ae4e8fbedcc4f000b9c52904f13ea9456379d373524dea7fef79b39f7","identifier":"test-composite","sealed":false,"timestamp":1522325758,"cipher":"aes-pmac-siv"}`,
+				1,
+			),
+			configstore.NewItem(
+				keyloader.EncryptionKeyConfigName,
+				`{"key":"QXdDW4N/jmJzpMu7i1zu4YF1opTn7H+eOk9CLFGBSFg=","identifier":"test-composite","sealed":false,"timestamp":1522325802,"cipher":"xchacha20-poly1305"}`,
+				1,
+			),
+		},
+	}
+	return ret, nil
+}
+
+func TestMain(m *testing.M) {
+	configstore.RegisterProvider("test", ProviderTest)
+	os.Exit(m.Run())
+}
+
+func TestIncompleteRead(t *testing.T) {
+	clearContent := make([]byte, 32*1024+10)
+	rand.Read(clearContent) // nolint
+
+	k, err := keyloader.LoadKey("test")
+	require.NoError(t, err)
+
+	var bufWriter bytes.Buffer
+	streamWriter := stream.NewWriter(&bufWriter, k, 32*1024)
+	nbBytesWritten, err := io.Copy(streamWriter, bytes.NewReader(clearContent))
+	require.NoError(t, err)
+	t.Logf("%d bytes copied to streamWriter", nbBytesWritten)
+	require.NoError(t, streamWriter.Close())
+
+	streamReader := stream.NewReader(strings.NewReader(bufWriter.String()), k, 32*1024)
+	var firstPart = make([]byte, 32*1024)
+	nbBytesReaden1, err := streamReader.Read(firstPart)
+	t.Logf("%d bytes read the first time", nbBytesReaden1)
+	require.NoError(t, err)
+
+	var secondPart = make([]byte, 32*1024)
+	nbBytesReaden2, err := streamReader.Read(secondPart)
+	t.Logf("%d bytes read the second time", nbBytesReaden2)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "EOF")
+
+	require.Equal(t, 32*1024+10, nbBytesReaden1+nbBytesReaden2)
+
+}