Verify blobs against the expected digests while copying them.

Note that this requires ImageDestination.PutBlob to fail and delete
any unfinished data if stream.Read() fails.

We do not have to trust PutBlob to correctly handle a validation error,
so we don't; but we can't do the storage cleanup for PutBlob.

cmd/skopeo/copy.go

@@ -1,8 +1,14 @@
 package main
 
 import (
+	"crypto/sha256"
+	"crypto/subtle"
+	"encoding/hex"
 	"errors"
 	"fmt"
+	"hash"
+	"io"
+	"strings"
 
 	"github.com/containers/image/image"
 	"github.com/containers/image/signature"
@@ -10,6 +16,65 @@ import (
 	"github.com/urfave/cli"
 )
 
+// supportedDigests lists the supported blob digest types.
+var supportedDigests = map[string]func() hash.Hash{
+	"sha256": sha256.New,
+}
+
+type digestingReader struct {
+	source           io.Reader
+	digest           hash.Hash
+	expectedDigest   []byte
+	failureIndicator *bool
+}
+
+// newDigestingReader returns an io.Reader with contents of source, which will eventually return a non-EOF error
+// and set *failureIndicator to true if the source stream does not match expectedDigestString.
+func newDigestingReader(source io.Reader, expectedDigestString string, failureIndicator *bool) (io.Reader, error) {
+	fields := strings.SplitN(expectedDigestString, ":", 2)
+	if len(fields) != 2 {
+		return nil, fmt.Errorf("Invalid digest specification %s", expectedDigestString)
+	}
+	fn, ok := supportedDigests[fields[0]]
+	if !ok {
+		return nil, fmt.Errorf("Invalid digest specification %s: unknown digest type %s", expectedDigestString, fields[0])
+	}
+	digest := fn()
+	expectedDigest, err := hex.DecodeString(fields[1])
+	if err != nil {
+		return nil, fmt.Errorf("Invalid digest value %s: %v", expectedDigestString, err)
+	}
+	if len(expectedDigest) != digest.Size() {
+		return nil, fmt.Errorf("Invalid digest specification %s: length %d does not match %d", expectedDigestString, len(expectedDigest), digest.Size())
+	}
+	return &digestingReader{
+		source:           source,
+		digest:           digest,
+		expectedDigest:   expectedDigest,
+		failureIndicator: failureIndicator,
+	}, nil
+}
+
+func (d *digestingReader) Read(p []byte) (int, error) {
+	n, err := d.source.Read(p)
+	if n > 0 {
+		if n2, err := d.digest.Write(p[:n]); n2 != n || err != nil {
+			// Coverage: This should not happen, the hash.Hash interface requires
+			// d.digest.Write to never return an error, and the io.Writer interface
+			// requires n2 == len(input) if no error is returned.
+			return 0, fmt.Errorf("Error updating digest during verification: %d vs. %d, %v", n2, n, err)
+		}
+	}
+	if err == io.EOF {
+		actualDigest := d.digest.Sum(nil)
+		if subtle.ConstantTimeCompare(actualDigest, d.expectedDigest) != 1 {
+			*d.failureIndicator = true
+			return 0, fmt.Errorf("Digest did not match, expected %s, got %s", hex.EncodeToString(d.expectedDigest), hex.EncodeToString(actualDigest))
+		}
+	}
+	return n, err
+}
+
 func copyHandler(context *cli.Context) error {
 	if len(context.Args()) != 2 {
 		return errors.New("Usage: copy source destination")
@@ -44,9 +109,23 @@ func copyHandler(context *cli.Context) error {
 			return fmt.Errorf("Error reading blob %s: %v", digest, err)
 		}
 		defer stream.Close()
-		if err := dest.PutBlob(digest, stream); err != nil {
+
+		// Be paranoid; in case PutBlob somehow managed to ignore an error from digestingReader,
+		// use a separate validation failure indicator.
+		// Note that we don't use a stronger "validationSucceeded" indicator, because
+		// dest.PutBlob may detect that the layer already exists, in which case we don't
+		// read stream to the end, and validation does not happen.
+		validationFailed := false // This is a new instance on each loop iteration.
+		digestingReader, err := newDigestingReader(stream, digest, &validationFailed)
+		if err != nil {
+			return fmt.Errorf("Error preparing to verify blob %s: %v", digest, err)
+		}
+		if err := dest.PutBlob(digest, digestingReader); err != nil {
 			return fmt.Errorf("Error writing blob: %v", err)
 		}
+		if validationFailed { // Coverage: This should never happen.
+			return fmt.Errorf("Internal error uploading blob %s, digest verification failed but was ignored", digest)
+		}
 	}
 
 	sigs, err := src.Signatures()

cmd/skopeo/copy_test.go

@@ -0,0 +1,62 @@
+package main
+
+import (
+	"bytes"
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewDigestingReader(t *testing.T) {
+	// Only the failure cases, success is tested in TestDigestingReaderRead below.
+	source := bytes.NewReader([]byte("abc"))
+	for _, input := range []string{
+		"abc",             // Not algo:hexvalue
+		"crc32:",          // Unknown algorithm, empty value
+		"crc32:012345678", // Unknown algorithm
+		"sha256:",         // Empty value
+		"sha256:0",        // Invalid hex value
+		"sha256:01",       // Invalid length of hex value
+	} {
+		validationFailed := false
+		_, err := newDigestingReader(source, input, &validationFailed)
+		assert.Error(t, err, input)
+	}
+}
+
+func TestDigestingReaderRead(t *testing.T) {
+	cases := []struct {
+		input  []byte
+		digest string
+	}{
+		{[]byte(""), "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"},
+		{[]byte("abc"), "sha256:ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"},
+		{make([]byte, 65537, 65537), "sha256:3266304f31be278d06c3bd3eb9aa3e00c59bedec0a890de466568b0b90b0e01f"},
+	}
+	// Valid input
+	for _, c := range cases {
+		source := bytes.NewReader(c.input)
+		validationFailed := false
+		reader, err := newDigestingReader(source, c.digest, &validationFailed)
+		require.NoError(t, err, c.digest)
+		dest := bytes.Buffer{}
+		n, err := io.Copy(&dest, reader)
+		assert.NoError(t, err, c.digest)
+		assert.Equal(t, int64(len(c.input)), n, c.digest)
+		assert.Equal(t, c.input, dest.Bytes(), c.digest)
+		assert.False(t, validationFailed, c.digest)
+	}
+	// Modified input
+	for _, c := range cases {
+		source := bytes.NewReader(bytes.Join([][]byte{c.input, []byte("x")}, nil))
+		validationFailed := false
+		reader, err := newDigestingReader(source, c.digest, &validationFailed)
+		require.NoError(t, err, c.digest)
+		dest := bytes.Buffer{}
+		_, err = io.Copy(&dest, reader)
+		assert.Error(t, err, c.digest)
+		assert.True(t, validationFailed)
+	}
+}