Pluggable blob storage for pkg/registry (#1209)

* Pluggable blob storage for pkg/registry

Features:
- supports redirects if the blob handler hosts blobs elsewhere.
- supports automatically verifying digests/sizes if possible,
  transparent to the blob handler (if they want it to be).
- repo strings passed to blobHandler methods aren't name.Repository, but
  should be easily parsed by implementations if they want that.

This is still hidden in unexported interfaces and always uses memHandler
for now, while we iterate on the shape we want these things to take.

Longer term, we should also extend this to manifest and upload storage,
using similar interfaces.

* Move blob access locks into memHandler

* pass context, unexport errors

* - split blobHandler into extensible interfaces
  - if the provided handler supports Stat, we use it; otherwise fallback to Get
  - if the provided handler doesn't support Put, the registry is read-only
- better registry error handling
  - reused vars (less verbose)
  - renamed errTODO(string) -> regErrInternal(err)
- added response body check to test cases
- checking verify.ReadCloser errors, verification is broken somehow (working on it)

* Fix verify.ReadCloser when size is unknown 🤦

* handle redirects for Stat, and fallback-Get on HEAD

* do type checks inline

internal/verify/verify.go

@@ -38,6 +38,18 @@ type verifyReader struct {
 	gotSize, wantSize int64
 }
 
+// Error provides information about the failed hash verification.
+type Error struct {
+	got     string
+	want    v1.Hash
+	gotSize int64
+}
+
+func (v Error) Error() string {
+	return fmt.Sprintf("error verifying %s checksum after reading %d bytes; got %q, want %q",
+		v.want.Algorithm, v.gotSize, v.got, v.want)
+}
+
 // Read implements io.Reader
 func (vc *verifyReader) Read(b []byte) (int, error) {
 	n, err := vc.inner.Read(b)
@@ -46,10 +58,13 @@ func (vc *verifyReader) Read(b []byte) (int, error) {
 		if vc.wantSize != SizeUnknown && vc.gotSize != vc.wantSize {
 			return n, fmt.Errorf("error verifying size; got %d, want %d", vc.gotSize, vc.wantSize)
 		}
-		got := hex.EncodeToString(vc.hasher.Sum(make([]byte, 0, vc.hasher.Size())))
+		got := hex.EncodeToString(vc.hasher.Sum(nil))
 		if want := vc.expected.Hex; got != want {
-			return n, fmt.Errorf("error verifying %s checksum after reading %d bytes; got %q, want %q",
-				vc.expected.Algorithm, vc.gotSize, got, want)
+			return n, Error{
+				got:     vc.expected.Algorithm + ":" + got,
+				want:    vc.expected,
+				gotSize: vc.gotSize,
+			}
 		}
 	}
 	return n, err
@@ -69,9 +84,9 @@ func ReadCloser(r io.ReadCloser, size int64, h v1.Hash) (io.ReadCloser, error) {
 	if err != nil {
 		return nil, err
 	}
-	var r2 io.Reader = r
+	r2 := io.TeeReader(r, w) // pass all writes to the hasher.
 	if size != SizeUnknown {
-		r2 = io.LimitReader(io.TeeReader(r, w), size)
+		r2 = io.LimitReader(r2, size) // if we know the size, limit to that size.
 	}
 	return &and.ReadCloser{
 		Reader: &verifyReader{

internal/verify/verify_test.go

@@ -30,6 +30,7 @@ func mustHash(s string, t *testing.T) v1.Hash {
 	if err != nil {
 		t.Fatalf("v1.SHA256(%s) = %v", s, err)
 	}
+	t.Logf("Hashed: %q -> %q", s, h)
 	return h
 }
 
@@ -59,6 +60,19 @@ func TestVerification(t *testing.T) {
 	}
 }
 
+func TestVerificationSizeUnknown(t *testing.T) {
+	want := "This is the input string."
+	buf := bytes.NewBufferString(want)
+
+	verified, err := ReadCloser(ioutil.NopCloser(buf), SizeUnknown, mustHash(want, t))
+	if err != nil {
+		t.Fatal("ReadCloser() =", err)
+	}
+	if _, err := ioutil.ReadAll(verified); err != nil {
+		t.Error("ReadAll() =", err)
+	}
+}
+
 func TestBadHash(t *testing.T) {
 	h := v1.Hash{
 		Algorithm: "fake256",