diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 4b7ebe72..f3fdcbb0 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -157,7 +157,9 @@ func (p *Proxy) ReadMetadata(r io.Reader) ([]byte, error) { // Reading into a buffer of that size avoids io.ReadAll growing its buffer // step by step, which peaks at more than twice the document size. size is // only a hint: a negative one, or one above the limit, reads as ReadMetadata -// does, and a body of a different length is still read correctly. +// does, and a body of a different length is still read correctly. A body +// read through upstreamBodyReader that upstream cuts off fails, even though +// it ends like a body shorter than size. func (p *Proxy) ReadMetadataSized(r io.Reader, size int64) ([]byte, error) { limit := p.MetadataMaxSize if limit <= 0 { @@ -170,7 +172,10 @@ func (p *Proxy) ReadMetadataSized(r io.Reader, size int64) ([]byte, error) { // the buffer and a longer one is noticed. buf := make([]byte, size+1) n, err := io.ReadFull(r, buf) + var cut *upstreamBodyError switch { + case errors.As(err, &cut): + return nil, err case errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF): if int64(n) < size/2 { // The hint was far off; don't keep the oversized buffer alive. @@ -2010,7 +2015,7 @@ func (p *Proxy) fetchUpstreamMetadata(ctx context.Context, upstreamURL string, e // ContentLength is -1 when the transport decompresses the body, and // otherwise the length of the bytes read here. - body, err := p.ReadMetadataSized(resp.Body, resp.ContentLength) + body, err := p.ReadMetadataSized(upstreamBodyReader{resp.Body}, resp.ContentLength) if err != nil { return nil, fmt.Errorf("reading response: %w", err) } diff --git a/internal/handler/metadata_truncated_test.go b/internal/handler/metadata_truncated_test.go new file mode 100644 index 00000000..50443745 --- /dev/null +++ b/internal/handler/metadata_truncated_test.go @@ -0,0 +1,121 @@ +package handler + +import ( + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/git-pkgs/cooldown" +) + +// cutMetadataUpstream serves one npm packument and, when cut is set, closes +// the connection halfway through the document it announced. +type cutMetadataUpstream struct { + *httptest.Server + mu sync.Mutex + body string + cut bool +} + +func newCutMetadataUpstream(t *testing.T, body string) *cutMetadataUpstream { + t.Helper() + u := &cutMetadataUpstream{body: body} + u.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + u.mu.Lock() + body, cut := u.body, u.cut + u.mu.Unlock() + if r.URL.Path != "/pkg" { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Content-Length", fmt.Sprint(len(body))) + if cut { + _, _ = io.WriteString(w, body[:len(body)/2]) + return + } + _, _ = io.WriteString(w, body) + })) + t.Cleanup(u.Close) + return u +} + +func (u *cutMetadataUpstream) set(body string, cut bool) { + u.mu.Lock() + defer u.mu.Unlock() + u.body, u.cut = body, cut +} + +// truncPackument is an npm packument with n versions published long ago. +func truncPackument(n int) string { + var versions, times []string + for i := range n { + v := fmt.Sprintf("1.0.%d", i) + versions = append(versions, fmt.Sprintf(`%q:{"name":"pkg","version":%q,"dist":{"tarball":"https://registry.npmjs.org/pkg/-/pkg-%s.tgz"}}`, v, v, v)) + times = append(times, fmt.Sprintf(`%q:"2020-01-01T00:00:00Z"`, v)) + } + return `{"name":"pkg","dist-tags":{"latest":"1.0.0"},"versions":{` + strings.Join(versions, ",") + + `},"time":{` + strings.Join(times, ",") + `}}` +} + +// TestMetadataTruncatedUpstream checks that a document upstream cuts off +// before its Content-Length is neither served nor cached: the cached copy is +// served instead, with and without cooldown. +func TestMetadataTruncatedUpstream(t *testing.T) { + for _, withCooldown := range []bool{false, true} { + t.Run(map[bool]string{false: "no cooldown", true: "cooldown"}[withCooldown], func(t *testing.T) { + u := newCutMetadataUpstream(t, truncPackument(50)) + proxy, db, _, _ := setupTestProxy(t) + proxy.HTTPClient = u.Client() + proxy.CacheMetadata = true + if withCooldown { + proxy.Cooldown = &cooldown.Config{Default: "3d"} + } + handler := NewNPMHandler(proxy, "http://proxy.example", u.URL).Routes() + + first := serve(handler, "/pkg", nil) + if first.Code != http.StatusOK { + t.Fatalf("first request: status %d", first.Code) + } + before, err := db.GetMetadataCache("npm", "pkg") + if err != nil || before == nil { + t.Fatalf("no metadata row after the first request: %v", err) + } + + u.set(truncPackument(51), true) + expireMetadata(t, db, "npm", "pkg") + proxy.SetMetadataRewriteCacheSize(0) + + rec := serve(handler, "/pkg", nil) + if rec.Code != http.StatusOK || rec.Body.String() != first.Body.String() { + t.Errorf("status %d, want 200 with the cached document", rec.Code) + } + after, err := db.GetMetadataCache("npm", "pkg") + if err != nil || after == nil || after.ContentDigest != before.ContentDigest { + t.Error("the cut-off document replaced the cached copy") + } + }) + } +} + +// TestMetadataTruncatedUpstreamUncached checks that a cut-off document with +// nothing cached fails the request and is not cached. +func TestMetadataTruncatedUpstreamUncached(t *testing.T) { + u := newCutMetadataUpstream(t, truncPackument(50)) + u.set(truncPackument(50), true) + proxy, db, _, _ := setupTestProxy(t) + proxy.HTTPClient = u.Client() + proxy.CacheMetadata = true + handler := NewNPMHandler(proxy, "http://proxy.example", u.URL).Routes() + + if rec := serve(handler, "/pkg", nil); rec.Code != http.StatusBadGateway { + t.Errorf("status %d, want 502", rec.Code) + } + if entry, _ := db.GetMetadataCache("npm", "pkg"); entry != nil { + t.Error("cached the cut-off document") + } +} diff --git a/internal/handler/read_metadata_test.go b/internal/handler/read_metadata_test.go index 639405c8..d87d4169 100644 --- a/internal/handler/read_metadata_test.go +++ b/internal/handler/read_metadata_test.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "runtime" + "strings" "testing" "testing/iotest" ) @@ -131,3 +132,18 @@ func TestReadMetadataSizedAllocatesOnce(t *testing.T) { t.Errorf("allocated %d bytes reading %d", allocated, size) } } + +// TestReadMetadataSizedTruncatedUpstream checks that an upstream body cut +// off before its Content-Length fails, while a body shorter than a size +// hint is still read. +func TestReadMetadataSizedTruncatedUpstream(t *testing.T) { + p := &Proxy{} + cut := io.MultiReader(strings.NewReader(`{"name":`), iotest.ErrReader(io.ErrUnexpectedEOF)) + if _, err := p.ReadMetadataSized(upstreamBodyReader{cut}, 100); !errors.Is(err, io.ErrUnexpectedEOF) { + t.Errorf("cut off body: error %v, want io.ErrUnexpectedEOF", err) + } + got, err := p.ReadMetadataSized(strings.NewReader(`{"name":"x"}`), 100) + if err != nil || string(got) != `{"name":"x"}` { + t.Errorf("short body = %q, %v", got, err) + } +} diff --git a/internal/handler/upstream_body.go b/internal/handler/upstream_body.go new file mode 100644 index 00000000..7db90765 --- /dev/null +++ b/internal/handler/upstream_body.go @@ -0,0 +1,26 @@ +package handler + +import ( + "errors" + "io" +) + +// upstreamBodyReader marks a failure to read a document from upstream, so a +// body upstream cuts off is told apart from one that ends, and from a +// failure to store it. +type upstreamBodyReader struct{ r io.Reader } + +func (u upstreamBodyReader) Read(b []byte) (int, error) { + n, err := u.r.Read(b) + if err != nil && !errors.Is(err, io.EOF) { + err = &upstreamBodyError{err} + } + return n, err +} + +// upstreamBodyError is a failure to read a document from upstream. It is +// handled as an upstream failure: the stored copy, if any, is served. +type upstreamBodyError struct{ err error } + +func (e *upstreamBodyError) Error() string { return "reading upstream metadata: " + e.err.Error() } +func (e *upstreamBodyError) Unwrap() error { return e.err }