diff --git a/src/controller/proxy/blobretry.go b/src/controller/proxy/blobretry.go new file mode 100644 index 00000000000..c03144617ab --- /dev/null +++ b/src/controller/proxy/blobretry.go @@ -0,0 +1,222 @@ +// Copyright Project Harbor Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package proxy + +import ( + "context" + "errors" + "fmt" + "io" + "time" + + "github.com/opencontainers/go-digest" + + libErrors "github.com/goharbor/harbor/src/lib/errors" + "github.com/goharbor/harbor/src/lib/log" +) + +const ( + // maxBlobFetchRetries bounds how many times an interrupted upstream blob + // fetch is resumed before the read error is returned to the caller. + maxBlobFetchRetries = 3 +) + +// blobFetchRetryBackoff is the delay before the first resume attempt; it +// doubles on each subsequent attempt (1s, 2s, 4s by default). It's a var +// rather than a const so tests can shrink it. +var blobFetchRetryBackoff = time.Second + +// resumeBlobFunc opens a reader for the remaining bytes of a blob, starting +// at offset. It's used to resume a fetch that broke mid-stream. +type resumeBlobFunc func(offset int64) (io.ReadCloser, error) + +// resumingBlobReader wraps an upstream blob body and transparently resumes +// the fetch, via resume, if the connection breaks mid-stream - instead of +// failing the whole blob on a single dropped connection, the way a plain +// docker/OCI client retries an interrupted layer pull. Resuming is bounded by +// maxBlobFetchRetries and disabled when size is unknown (<= 0), since the +// read offset can't be validated against the blob length in that case. +// +// resume itself isn't cancelable once a request is in flight - the +// underlying RemoteInterface doesn't accept a context - but the backoff wait +// between attempts, and any attempt not yet started, respect ctx, so a +// canceled caller (e.g. a disconnected pulling client) stops this reader +// from waiting out further backoffs and opening further upstream requests. +type resumingBlobReader struct { + ctx context.Context + reader io.ReadCloser + resume resumeBlobFunc + size int64 + offset int64 + retries int + completed bool +} + +// newResumingBlobReader wraps reader so a mid-stream read failure is retried +// up to maxBlobFetchRetries times by resuming the fetch from the last +// successfully read byte offset (via an HTTP range request through resume). +// ctx bounds the retry loop itself (see resumingBlobReader); pass the +// pulling client's request context to give up promptly if it disconnects, +// or a detached context (e.g. context.Background()) for background work +// that should keep retrying regardless. +func newResumingBlobReader(ctx context.Context, reader io.ReadCloser, size int64, resume resumeBlobFunc) io.ReadCloser { + return &resumingBlobReader{ctx: ctx, reader: reader, size: size, resume: resume} +} + +func (r *resumingBlobReader) Read(p []byte) (int, error) { + if r.completed { + // A prior call already reached the declared size; self-terminate + // instead of calling into a reader that may now be stale/exhausted. + return 0, io.EOF + } + for { + n, err := r.reader.Read(p) + r.offset += int64(n) + + if r.size > 0 && r.offset >= r.size { + // We now have the full declared size, regardless of what error + // (if any) came with these final bytes - an io.Reader may + // legally pair its last bytes with a non-nil error instead of + // returning them cleanly and erroring only on a subsequent + // call. Report success once; further calls are handled by the + // completed check above rather than re-reading this reader. + r.completed = true + return n, nil + } + + if err == nil { + return n, nil + } + if err == io.EOF { // nolint:errorlint + if r.size <= 0 { + // Nothing to validate the EOF against; trust it. + return n, err + } + // The stream ended cleanly but short of the expected size - + // e.g. a network intermediary that terminates a chunked + // response gracefully instead of resetting the connection. + // Treat it the same as a dropped connection so it consumes + // the retry budget instead of being accepted as complete. + err = io.ErrUnexpectedEOF + } + if !r.canRetry(err) { + return n, err + } + if rErr := r.reconnect(err); rErr != nil { + return n, rErr + } + if n > 0 { + return n, nil + } + // n == 0: loop and read immediately from the freshly resumed reader. + } +} + +// reconnect closes the broken reader and resumes the fetch from the current +// offset, retrying the resume itself (bounded by the same retry budget) if +// establishing the new connection also fails. It gives up early, without +// waiting out the backoff or starting another attempt, if ctx is canceled. +func (r *resumingBlobReader) reconnect(cause error) error { + _ = r.reader.Close() + for { + if ctxErr := r.ctx.Err(); ctxErr != nil { + return fmt.Errorf("blob fetch canceled while resuming after %v: %w", cause, ctxErr) + } + r.retries++ + backoff := blobFetchRetryBackoff * time.Duration(uint64(1)<= maxBlobFetchRetries { + return fmt.Errorf("failed to resume interrupted blob fetch after %v: %w", cause, err) + } + cause = err + } +} + +func (r *resumingBlobReader) canRetry(err error) bool { + return r.size > 0 && r.retries < maxBlobFetchRetries && r.offset < r.size && isRetryableBlobReadErr(err) +} + +func (r *resumingBlobReader) Close() error { + return r.reader.Close() +} + +// isRetryableBlobReadErr reports whether err looks like a transient failure +// of the upstream connection - dropped, reset, or timed out mid-transfer - +// worth retrying, as opposed to the pulling client itself having given up. +func isRetryableBlobReadErr(err error) bool { + return err != nil && !errors.Is(err, context.Canceled) +} + +// verifyingReadCloser wraps a blob reader and verifies the bytes read so far +// match the expected digest as soon as the declared size has been read - not +// only when the wrapped reader happens to signal io.EOF on some later call. +// A consumer with its own framing (e.g. an HTTP request body capped by +// Content-Length) may never issue that extra call, since it can determine +// on its own that it has all the bytes it declared and stop reading. A +// mismatch is surfaced as a Read error, so a corrupted or truncated blob +// that still parses as a well-formed HTTP response is never mistaken for a +// complete, valid one by the caller (e.g. the local registry committing a +// blob push). +type verifyingReadCloser struct { + io.ReadCloser + verifier digest.Verifier + dig digest.Digest + size int64 + read int64 + verified bool +} + +// newVerifyingReadCloser wraps reader so that, once size bytes have been +// read (or the stream ends, if size is unknown), content not matching dig +// surfaces as a Read error. +func newVerifyingReadCloser(reader io.ReadCloser, dig digest.Digest, size int64) io.ReadCloser { + return &verifyingReadCloser{ReadCloser: reader, verifier: dig.Verifier(), dig: dig, size: size} +} + +func (v *verifyingReadCloser) Read(p []byte) (int, error) { + n, err := v.ReadCloser.Read(p) + if n > 0 { + _, _ = v.verifier.Write(p[:n]) + v.read += int64(n) + } + if !v.verified && ((v.size > 0 && v.read >= v.size) || err == io.EOF) { // nolint:errorlint + v.verified = true + if !v.verifier.Verified() { + return n, fmt.Errorf("blob content does not match expected digest %s after fetching from upstream", v.dig) + } + } + return n, err +} diff --git a/src/controller/proxy/blobretry_test.go b/src/controller/proxy/blobretry_test.go new file mode 100644 index 00000000000..d10cc1365c2 --- /dev/null +++ b/src/controller/proxy/blobretry_test.go @@ -0,0 +1,271 @@ +// Copyright Project Harbor Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package proxy + +import ( + "bytes" + "context" + "errors" + "io" + "testing" + "time" + + "github.com/opencontainers/go-digest" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + libErrors "github.com/goharbor/harbor/src/lib/errors" +) + +// stubReader serves data and then, once exhausted, returns err (or io.EOF if +// err is nil). It records whether it was closed. +type stubReader struct { + data []byte + err error + closed bool +} + +func (s *stubReader) Read(p []byte) (int, error) { + if len(s.data) > 0 { + n := copy(p, s.data) + s.data = s.data[n:] + return n, nil + } + if s.err != nil { + return 0, s.err + } + return 0, io.EOF +} + +func (s *stubReader) Close() error { + s.closed = true + return nil +} + +// oneShotReader returns all of data together with err in a single Read call, +// unlike stubReader, which always separates the final data from a +// subsequent terminal error/EOF call - both are legal io.Reader behavior. +type oneShotReader struct { + data []byte + err error +} + +func (o *oneShotReader) Read(p []byte) (int, error) { + n := copy(p, o.data) + o.data = o.data[n:] + return n, o.err +} + +func (o *oneShotReader) Close() error { return nil } + +func withShortBlobRetryBackoff(t *testing.T) { + orig := blobFetchRetryBackoff + blobFetchRetryBackoff = time.Millisecond + t.Cleanup(func() { blobFetchRetryBackoff = orig }) +} + +func TestResumingBlobReader_ResumesAfterMidStreamError(t *testing.T) { + withShortBlobRetryBackoff(t) + + first := &stubReader{data: []byte("hello "), err: io.ErrUnexpectedEOF} + second := &stubReader{data: []byte("world")} + + var resumeOffsets []int64 + reader := newResumingBlobReader(context.Background(), first, 11, func(offset int64) (io.ReadCloser, error) { + resumeOffsets = append(resumeOffsets, offset) + return second, nil + }) + + got, err := io.ReadAll(reader) + require.NoError(t, err) + assert.Equal(t, "hello world", string(got)) + assert.Equal(t, []int64{6}, resumeOffsets) + assert.True(t, first.closed, "the broken reader should be closed before resuming") +} + +func TestResumingBlobReader_GivesUpAfterMaxRetries(t *testing.T) { + withShortBlobRetryBackoff(t) + + first := &stubReader{err: io.ErrUnexpectedEOF} + resumeAttempts := 0 + reader := newResumingBlobReader(context.Background(), first, 100, func(offset int64) (io.ReadCloser, error) { + resumeAttempts++ + return nil, errors.New("upstream still unreachable") + }) + + _, err := io.ReadAll(reader) + require.Error(t, err) + assert.Equal(t, maxBlobFetchRetries, resumeAttempts) +} + +func TestResumingBlobReader_GivesUpImmediatelyOnRateLimitedResume(t *testing.T) { + withShortBlobRetryBackoff(t) + + first := &stubReader{err: io.ErrUnexpectedEOF} + resumeAttempts := 0 + rateLimitErr := libErrors.New("too many requests").WithCode(libErrors.RateLimitCode) + reader := newResumingBlobReader(context.Background(), first, 100, func(offset int64) (io.ReadCloser, error) { + resumeAttempts++ + return nil, rateLimitErr + }) + + _, err := io.ReadAll(reader) + require.Error(t, err) + assert.Equal(t, 1, resumeAttempts, "a rate-limited resume shouldn't be retried - a few seconds of backoff won't help and it just adds more load") +} + +func TestResumingBlobReader_DoesNotRetryOnContextCanceled(t *testing.T) { + withShortBlobRetryBackoff(t) + + first := &stubReader{err: context.Canceled} + resumeAttempts := 0 + reader := newResumingBlobReader(context.Background(), first, 100, func(offset int64) (io.ReadCloser, error) { + resumeAttempts++ + return &stubReader{}, nil + }) + + _, err := io.ReadAll(reader) + require.Error(t, err) + assert.Equal(t, 0, resumeAttempts, "a canceled request should not be retried") +} + +func TestResumingBlobReader_StopsRetryingWhenContextCanceled(t *testing.T) { + withShortBlobRetryBackoff(t) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // already canceled before the reader ever attempts to resume + + first := &stubReader{err: io.ErrUnexpectedEOF} + resumeAttempts := 0 + reader := newResumingBlobReader(ctx, first, 100, func(offset int64) (io.ReadCloser, error) { + resumeAttempts++ + return &stubReader{}, nil + }) + + _, err := io.ReadAll(reader) + require.Error(t, err) + assert.Equal(t, 0, resumeAttempts, "a canceled context should stop retrying before opening another upstream request") +} + +func TestResumingBlobReader_DoesNotRetryWithUnknownSize(t *testing.T) { + withShortBlobRetryBackoff(t) + + first := &stubReader{err: io.ErrUnexpectedEOF} + resumeAttempts := 0 + reader := newResumingBlobReader(context.Background(), first, 0, func(offset int64) (io.ReadCloser, error) { + resumeAttempts++ + return &stubReader{}, nil + }) + + _, err := io.ReadAll(reader) + require.Error(t, err) + assert.Equal(t, 0, resumeAttempts, "resuming an unknown-length blob can't be validated, so it shouldn't be attempted") +} + +func TestResumingBlobReader_RecoversFromMultipleInterruptions(t *testing.T) { + withShortBlobRetryBackoff(t) + + first := &stubReader{data: []byte("aaa"), err: io.ErrUnexpectedEOF} + second := &stubReader{data: []byte("bbb"), err: io.ErrUnexpectedEOF} + third := &stubReader{data: []byte("ccc")} + var resumeOffsets []int64 + readers := []*stubReader{second, third} + reader := newResumingBlobReader(context.Background(), first, 9, func(offset int64) (io.ReadCloser, error) { + resumeOffsets = append(resumeOffsets, offset) + next := readers[0] + readers = readers[1:] + return next, nil + }) + + got, err := io.ReadAll(reader) + require.NoError(t, err) + assert.Equal(t, "aaabbbccc", string(got)) + assert.Equal(t, []int64{3, 6}, resumeOffsets) +} + +func TestResumingBlobReader_RetriesOnPrematureCleanEOF(t *testing.T) { + withShortBlobRetryBackoff(t) + + // first ends with a clean nil error (io.EOF on the next call), not a + // transport error - e.g. a network intermediary that terminates a + // chunked response gracefully instead of resetting the connection. + first := &stubReader{data: []byte("hello ")} + second := &stubReader{data: []byte("world")} + + var resumeOffsets []int64 + reader := newResumingBlobReader(context.Background(), first, 11, func(offset int64) (io.ReadCloser, error) { + resumeOffsets = append(resumeOffsets, offset) + return second, nil + }) + + got, err := io.ReadAll(reader) + require.NoError(t, err) + assert.Equal(t, "hello world", string(got)) + assert.Equal(t, []int64{6}, resumeOffsets, "a clean EOF short of the declared size must still trigger a resume") +} + +func TestResumingBlobReader_TreatsErrorAccompanyingFinalBytesAsSuccess(t *testing.T) { + // An io.Reader may legally return its final bytes together with a + // non-nil, non-EOF error in the same call, instead of returning them + // cleanly and erroring only on a subsequent call. If those bytes bring + // the reader to the full declared size, that's a complete blob and must + // not be treated as a failed/retryable read. + content := []byte("complete blob") + reader := newResumingBlobReader(context.Background(), &oneShotReader{data: content, err: io.ErrUnexpectedEOF}, int64(len(content)), + func(offset int64) (io.ReadCloser, error) { + t.Fatal("resume should not be called when the blob was already fully read") + return nil, nil + }) + + got, err := io.ReadAll(reader) + require.NoError(t, err) + assert.Equal(t, content, got) +} + +func TestVerifyingReadCloser_PassesThroughValidContent(t *testing.T) { + content := []byte("valid blob content") + dig := digest.FromBytes(content) + + reader := newVerifyingReadCloser(io.NopCloser(&stubReader{data: content}), dig, int64(len(content))) + got, err := io.ReadAll(reader) + require.NoError(t, err) + assert.Equal(t, content, got) +} + +func TestVerifyingReadCloser_RejectsMismatchedContent(t *testing.T) { + content := []byte("corrupted blob content") + wrongDigest := digest.FromBytes([]byte("something else entirely")) + + reader := newVerifyingReadCloser(io.NopCloser(&stubReader{data: content}), wrongDigest, int64(len(content))) + _, err := io.ReadAll(reader) + require.Error(t, err) + assert.Contains(t, err.Error(), "does not match expected digest") +} + +func TestVerifyingReadCloser_VerifiesAsSoonAsSizeReachedWithoutFurtherEOFCall(t *testing.T) { + // bytes.Reader returns the final chunk with a nil error and only signals + // io.EOF on a subsequent call - the same pattern net/http's + // io.LimitReader-wrapped request body uses. A consumer capped by + // Content-Length (e.g. the real blob upload) may never make that + // subsequent call, so verification must not depend on it. + content := []byte("AAAAAAAAAA") + wrongDigest := digest.FromBytes([]byte("BBBBBBBBBB")) + + reader := newVerifyingReadCloser(io.NopCloser(bytes.NewReader(content)), wrongDigest, int64(len(content))) + buf := make([]byte, len(content)) + n, err := reader.Read(buf) + require.Error(t, err, "the read call that reaches the declared size must itself surface the mismatch") + assert.Equal(t, len(content), n) +} diff --git a/src/controller/proxy/controller.go b/src/controller/proxy/controller.go index efff91a7317..e5e1985f6c8 100644 --- a/src/controller/proxy/controller.go +++ b/src/controller/proxy/controller.go @@ -295,6 +295,14 @@ func (c *controller) ProxyBlob(ctx context.Context, p *proModels.Project, art li log.Errorf("failed to pull blob, error %v", err) return 0, nil, err } + // Resume the fetch from the last read byte offset if the upstream + // connection breaks mid-stream, instead of failing the pull outright. + // Bound retries to the request context so a disconnected pulling client + // stops this from waiting out backoffs and opening further upstream + // requests. + bReader = newResumingBlobReader(ctx, bReader, size, func(offset int64) (io.ReadCloser, error) { + return rHelper.BlobReaderAt(remoteRepo, art.Digest, size, offset) + }) desc := distribution.Descriptor{Size: size, Digest: digest.Digest(art.Digest)} go func() { err := c.putBlobToLocal(remoteRepo, art.Repository, desc, rHelper) @@ -312,6 +320,17 @@ func (c *controller) putBlobToLocal(remoteRepo string, localRepo string, desc di log.Errorf("failed to create blob reader, error %v", err) return err } + // Resume on a dropped upstream connection, same as the client-facing + // fetch above, and verify the assembled content against the expected + // digest so a corrupted/truncated blob that still parses as a valid HTTP + // response is never pushed to local storage as if it were complete. This + // runs in a background goroutine detached from the pulling client's + // request, so it uses its own context rather than inheriting one that + // would be canceled the moment that client disconnects. + bReader = newResumingBlobReader(context.Background(), bReader, desc.Size, func(offset int64) (io.ReadCloser, error) { + return r.BlobReaderAt(remoteRepo, string(desc.Digest), desc.Size, offset) + }) + bReader = newVerifyingReadCloser(bReader, desc.Digest, desc.Size) defer bReader.Close() err = c.local.PushBlob(localRepo, desc, bReader) return err diff --git a/src/controller/proxy/controller_test.go b/src/controller/proxy/controller_test.go index 9a3fe7f0236..a64a1ca95da 100644 --- a/src/controller/proxy/controller_test.go +++ b/src/controller/proxy/controller_test.go @@ -60,8 +60,11 @@ func (l *localInterfaceMock) BlobExist(ctx context.Context, art lib.ArtifactInfo return args.Bool(0), args.Error(1) } -func (l *localInterfaceMock) PushBlob(localRepo string, desc distribution.Descriptor, bReader io.ReadCloser) error { - panic("implement me") +func (l *localInterfaceMock) PushBlob(_ string, _ distribution.Descriptor, bReader io.ReadCloser) error { + // Drain the reader the way the real registry client would, so tests can + // exercise the retry/digest-verification wrapping around the reader chain. + _, err := io.ReadAll(bReader) + return err } func (l *localInterfaceMock) PushManifest(repo string, tag string, manifest distribution.Manifest) error { @@ -202,6 +205,34 @@ func (p *proxyControllerTestSuite) TestUseLocalBlob_False() { p.Assert().False(result) } +func (p *proxyControllerTestSuite) TestPutBlobToLocal_ResumesAfterInterruptionAndVerifiesDigest() { + withShortBlobRetryBackoff(p.T()) + content := []byte("hello dependency blob") + dig := digest.FromBytes(content) + desc := distribution.Descriptor{Size: int64(len(content)), Digest: dig} + + first := &stubReader{data: content[:5], err: io.ErrUnexpectedEOF} + second := &stubReader{data: content[5:]} + p.remote.On("BlobReader", "library/hello-world", string(dig)).Return(desc.Size, io.ReadCloser(first), nil) + p.remote.On("BlobReaderAt", "library/hello-world", string(dig), desc.Size, int64(5)).Return(io.ReadCloser(second), nil) + + err := p.ctr.(*controller).putBlobToLocal("library/hello-world", "proxy/library/hello-world", desc, p.remote) + p.Assert().NoError(err) + p.remote.AssertExpectations(p.T()) +} + +func (p *proxyControllerTestSuite) TestPutBlobToLocal_RejectsDigestMismatch() { + content := []byte("actual upstream content") + wrongDigest := digest.FromBytes([]byte("a different blob entirely")) + desc := distribution.Descriptor{Size: int64(len(content)), Digest: wrongDigest} + + p.remote.On("BlobReader", "library/hello-world", string(wrongDigest)).Return(desc.Size, io.ReadCloser(&stubReader{data: content}), nil) + + err := p.ctr.(*controller).putBlobToLocal("library/hello-world", "proxy/library/hello-world", desc, p.remote) + p.Assert().Error(err) + p.remote.AssertExpectations(p.T()) +} + func TestProxyControllerTestSuite(t *testing.T) { suite.Run(t, &proxyControllerTestSuite{}) } diff --git a/src/controller/proxy/manifestcache.go b/src/controller/proxy/manifestcache.go index f3a40ad339c..56528caccf5 100644 --- a/src/controller/proxy/manifestcache.go +++ b/src/controller/proxy/manifestcache.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "io" "strings" "time" @@ -244,6 +245,15 @@ func (m *ManifestCache) putBlobToLocal(remoteRepo string, localRepo string, desc log.Errorf("failed to create blob reader, error %v", err) return err } + // Resume on a dropped upstream connection and verify the assembled + // content against the expected digest, same as controller.putBlobToLocal, + // so a corrupted/truncated blob is never pushed to local storage as if + // it were complete. This is background work detached from any pulling + // client's request, so it uses its own context. + bReader = newResumingBlobReader(context.Background(), bReader, desc.Size, func(offset int64) (io.ReadCloser, error) { + return r.BlobReaderAt(remoteRepo, string(desc.Digest), desc.Size, offset) + }) + bReader = newVerifyingReadCloser(bReader, desc.Digest, desc.Size) defer bReader.Close() err = m.local.PushBlob(localRepo, desc, bReader) return err diff --git a/src/controller/proxy/manifestcache_test.go b/src/controller/proxy/manifestcache_test.go index cf54e043d0e..72ae67e9373 100644 --- a/src/controller/proxy/manifestcache_test.go +++ b/src/controller/proxy/manifestcache_test.go @@ -17,6 +17,7 @@ package proxy import ( "context" "fmt" + "io" "testing" "github.com/docker/distribution" @@ -28,6 +29,7 @@ import ( "github.com/goharbor/harbor/src/controller/artifact" "github.com/goharbor/harbor/src/lib" + testproxy "github.com/goharbor/harbor/src/testing/controller/proxy" "github.com/goharbor/harbor/src/testing/mock" v1 "github.com/opencontainers/image-spec/specs-go/v1" @@ -251,6 +253,36 @@ func (suite *CacheTestSuite) TestManifestCache_push_fails() { suite.Assert().Contains(errs, tagErr) } +func (suite *CacheTestSuite) TestManifestCache_putBlobToLocal_ResumesAfterInterruption() { + withShortBlobRetryBackoff(suite.T()) + content := []byte("dependency blob content") + dig := digest.FromBytes(content) + desc := distribution.Descriptor{Size: int64(len(content)), Digest: dig} + + first := &stubReader{data: content[:7], err: io.ErrUnexpectedEOF} + second := &stubReader{data: content[7:]} + remote := &testproxy.RemoteInterface{} + remote.On("BlobReader", "library/hello-world", string(dig)).Return(desc.Size, io.ReadCloser(first), nil) + remote.On("BlobReaderAt", "library/hello-world", string(dig), desc.Size, int64(7)).Return(io.ReadCloser(second), nil) + + err := suite.mCache.putBlobToLocal("library/hello-world", "proxy/library/hello-world", desc, remote) + suite.Assert().NoError(err) + remote.AssertExpectations(suite.T()) +} + +func (suite *CacheTestSuite) TestManifestCache_putBlobToLocal_RejectsDigestMismatch() { + content := []byte("dependency blob content") + wrongDigest := digest.FromBytes([]byte("something else")) + desc := distribution.Descriptor{Size: int64(len(content)), Digest: wrongDigest} + + remote := &testproxy.RemoteInterface{} + remote.On("BlobReader", "library/hello-world", string(wrongDigest)).Return(desc.Size, io.ReadCloser(&stubReader{data: content}), nil) + + err := suite.mCache.putBlobToLocal("library/hello-world", "proxy/library/hello-world", desc, remote) + suite.Assert().Error(err) + remote.AssertExpectations(suite.T()) +} + func TestCacheTestSuite(t *testing.T) { suite.Run(t, &CacheTestSuite{}) } diff --git a/src/controller/proxy/remote.go b/src/controller/proxy/remote.go index 781f19a13db..e36ce212642 100644 --- a/src/controller/proxy/remote.go +++ b/src/controller/proxy/remote.go @@ -32,6 +32,10 @@ import ( type RemoteInterface interface { // BlobReader create a reader for remote blob BlobReader(repo, dig string) (int64, io.ReadCloser, error) + // BlobReaderAt creates a reader for the remaining bytes of a remote blob, + // starting at the given offset (via an HTTP range request). It's used to + // resume a blob fetch that was interrupted mid-stream. + BlobReaderAt(repo, dig string, size, start int64) (io.ReadCloser, error) // Manifest get manifest by reference Manifest(repo string, ref string) (distribution.Manifest, string, error) // ManifestExist checks manifest exist, if exist, returns digest @@ -100,6 +104,17 @@ func (r *remoteHelper) BlobReader(repo, dig string) (int64, io.ReadCloser, error return sz, bReader, err } +func (r *remoteHelper) BlobReaderAt(repo, dig string, size, start int64) (io.ReadCloser, error) { + _, bReader, err := r.registry.PullBlobChunk(repo, dig, size, start, size-1) + if err != nil { + return nil, err + } + if r.opts != nil && r.opts.Speed > 0 { + bReader = lib.NewReader(bReader, r.opts.Speed) + } + return bReader, nil +} + func (r *remoteHelper) Manifest(repo string, ref string) (distribution.Manifest, string, error) { return r.registry.PullManifest(repo, ref) } diff --git a/src/pkg/registry/client.go b/src/pkg/registry/client.go index 0b59b5582cb..eb19c1f8274 100644 --- a/src/pkg/registry/client.go +++ b/src/pkg/registry/client.go @@ -385,7 +385,7 @@ func (c *client) PullBlob(repository, digest string) (int64, io.ReadCloser, erro } // PullBlobChunk pulls the specified blob, but by chunked, refer to https://github.com/opencontainers/distribution-spec/blob/main/spec.md#pull for more details. -func (c *client) PullBlobChunk(repository, digest string, _ int64, start, end int64) (int64, io.ReadCloser, error) { +func (c *client) PullBlobChunk(repository, digest string, blobSize, start, end int64) (int64, io.ReadCloser, error) { req, err := http.NewRequest(http.MethodGet, buildBlobURL(c.url, repository, digest), nil) if err != nil { return 0, nil, err @@ -398,6 +398,38 @@ func (c *client) PullBlobChunk(repository, digest string, _ int64, start, end in return 0, nil, err } + // A server is allowed to ignore the Range header and return the full blob + // with a 200 instead of a 206. Callers use this method to resume a fetch + // from a byte offset, so silently accepting a mismatched range here would + // splice bytes from the wrong offset into whatever they're assembling. + // Require a 206 whose Content-Range actually starts at the requested + // offset, or a plain 200 only when the whole blob was requested from the + // start. + switch resp.StatusCode { + case http.StatusPartialContent: + cr := resp.Header.Get("Content-Range") + got, crErr := parseContentRangeStart(cr) + if crErr != nil || got != start { + defer resp.Body.Close() + return 0, nil, fmt.Errorf("registry returned Content-Range %q for requested range bytes=%d-%d, refusing to treat it as that chunk", cr, start, end) + } + case http.StatusOK: + // A 200 only actually satisfies this request if the whole blob was + // requested (start at 0 through the last byte); otherwise - e.g. the + // first of several fixed-size chunks of a larger blob - a full body + // is not the same as the requested sub-range, even though both start + // at 0. When blobSize isn't known, fall back to the weaker start==0 + // check, since there's nothing to validate end against. + wholeBlobRequested := start == 0 && (blobSize <= 0 || end == blobSize-1) + if !wholeBlobRequested { + defer resp.Body.Close() + return 0, nil, fmt.Errorf("registry ignored range request bytes=%d-%d and returned a full 200 response", start, end) + } + default: + defer resp.Body.Close() + return 0, nil, fmt.Errorf("unexpected status code %d for ranged blob pull", resp.StatusCode) + } + var size int64 n := resp.Header.Get("Content-Length") // no content-length is acceptable, which can taken from manifests @@ -412,6 +444,17 @@ func (c *client) PullBlobChunk(repository, digest string, _ int64, start, end in return size, resp.Body, nil } +// parseContentRangeStart extracts the start offset from a GET response's +// Content-Range header, e.g. "bytes 100-499/1234" or "bytes 100-499/*". +func parseContentRangeStart(cr string) (int64, error) { + cr = strings.TrimPrefix(cr, "bytes ") + dash := strings.IndexByte(cr, '-') + if dash <= 0 { + return -1, fmt.Errorf("invalid Content-Range format: %q", cr) + } + return strconv.ParseInt(cr[:dash], 10, 64) +} + func (c *client) PushBlob(repository, digest string, size int64, blob io.Reader) error { location, _, err := c.initiateBlobUpload(repository) if err != nil { diff --git a/src/pkg/registry/client_test.go b/src/pkg/registry/client_test.go index 5bde2325bdc..739615831f1 100644 --- a/src/pkg/registry/client_test.go +++ b/src/pkg/registry/client_test.go @@ -325,6 +325,119 @@ func (c *clientTestSuite) TestPullBlob() { c.EqualValues(data, b) } +func (c *clientTestSuite) TestPullBlobChunk_PartialContentMatchingRange() { + data := []byte("world") + server := test.NewServer( + &test.RequestHandlerMapping{ + Method: "GET", + Pattern: "/v2/library/hello-world/blobs/digest", + Handler: test.Handler(&test.Response{ + StatusCode: http.StatusPartialContent, + Headers: map[string]string{ + "Content-Length": strconv.Itoa(len(data)), + "Content-Range": "bytes 6-10/11", + }, + Body: data, + }), + }) + defer server.Close() + + size, blob, err := NewClient(server.URL, "", "", true).PullBlobChunk("library/hello-world", "digest", 11, 6, 10) + c.Require().NoError(err) + c.Equal(int64(len(data)), size) + b, err := io.ReadAll(blob) + c.Require().NoError(err) + c.EqualValues(data, b) +} + +func (c *clientTestSuite) TestPullBlobChunk_FullContentAtZeroOffset() { + data := []byte("hello world") + server := test.NewServer( + &test.RequestHandlerMapping{ + Method: "GET", + Pattern: "/v2/library/hello-world/blobs/digest", + Handler: test.Handler(&test.Response{ + StatusCode: http.StatusOK, + Headers: map[string]string{ + "Content-Length": strconv.Itoa(len(data)), + }, + Body: data, + }), + }) + defer server.Close() + + size, blob, err := NewClient(server.URL, "", "", true).PullBlobChunk("library/hello-world", "digest", 11, 0, 10) + c.Require().NoError(err) + c.Equal(int64(len(data)), size) + b, err := io.ReadAll(blob) + c.Require().NoError(err) + c.EqualValues(data, b) +} + +func (c *clientTestSuite) TestPullBlobChunk_RejectsIgnoredRangeAtNonZeroOffset() { + data := []byte("hello world") + server := test.NewServer( + &test.RequestHandlerMapping{ + Method: "GET", + Pattern: "/v2/library/hello-world/blobs/digest", + Handler: test.Handler(&test.Response{ + StatusCode: http.StatusOK, + Headers: map[string]string{ + "Content-Length": strconv.Itoa(len(data)), + }, + Body: data, + }), + }) + defer server.Close() + + _, _, err := NewClient(server.URL, "", "", true).PullBlobChunk("library/hello-world", "digest", 11, 6, 10) + c.Require().Error(err) +} + +func (c *clientTestSuite) TestPullBlobChunk_RejectsFull200ForPartialFirstChunk() { + // start == 0 but end is short of blobSize-1, e.g. the first of several + // fixed-size chunks of a larger blob - a full 200 body is not the same + // as that sub-range even though both start at byte 0. + data := []byte("hello world") + server := test.NewServer( + &test.RequestHandlerMapping{ + Method: "GET", + Pattern: "/v2/library/hello-world/blobs/digest", + Handler: test.Handler(&test.Response{ + StatusCode: http.StatusOK, + Headers: map[string]string{ + "Content-Length": strconv.Itoa(len(data)), + }, + Body: data, + }), + }) + defer server.Close() + + _, _, err := NewClient(server.URL, "", "", true).PullBlobChunk("library/hello-world", "digest", int64(len(data)+50), 0, 4) + c.Require().Error(err) +} + +func (c *clientTestSuite) TestPullBlobChunk_RejectsMismatchedContentRange() { + data := []byte("wrong chunk") + server := test.NewServer( + &test.RequestHandlerMapping{ + Method: "GET", + Pattern: "/v2/library/hello-world/blobs/digest", + Handler: test.Handler(&test.Response{ + StatusCode: http.StatusPartialContent, + Headers: map[string]string{ + "Content-Length": strconv.Itoa(len(data)), + "Content-Range": "bytes 0-10/11", + }, + Body: data, + }), + }) + defer server.Close() + + _, _, err := NewClient(server.URL, "", "", true).PullBlobChunk("library/hello-world", "digest", 11, 6, 10) + c.Require().Error(err) +} + func (c *clientTestSuite) TestPushBlob() { server := test.NewServer( &test.RequestHandlerMapping{ diff --git a/src/testing/controller/proxy/remote_interface.go b/src/testing/controller/proxy/remote_interface.go index fffd363118e..7ed1dd9b5dc 100644 --- a/src/testing/controller/proxy/remote_interface.go +++ b/src/testing/controller/proxy/remote_interface.go @@ -53,6 +53,36 @@ func (_m *RemoteInterface) BlobReader(repo string, dig string) (int64, io.ReadCl return r0, r1, r2 } +// BlobReaderAt provides a mock function with given fields: repo, dig, size, start +func (_m *RemoteInterface) BlobReaderAt(repo string, dig string, size int64, start int64) (io.ReadCloser, error) { + ret := _m.Called(repo, dig, size, start) + + if len(ret) == 0 { + panic("no return value specified for BlobReaderAt") + } + + var r0 io.ReadCloser + var r1 error + if rf, ok := ret.Get(0).(func(string, string, int64, int64) (io.ReadCloser, error)); ok { + return rf(repo, dig, size, start) + } + if rf, ok := ret.Get(0).(func(string, string, int64, int64) io.ReadCloser); ok { + r0 = rf(repo, dig, size, start) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(io.ReadCloser) + } + } + + if rf, ok := ret.Get(1).(func(string, string, int64, int64) error); ok { + r1 = rf(repo, dig, size, start) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // ListReferrers provides a mock function with given fields: repo, digest, rawQuery func (_m *RemoteInterface) ListReferrers(repo string, digest string, rawQuery string) (*v1.Index, map[string][]string, error) { ret := _m.Called(repo, digest, rawQuery)