From 558dae0c3cf17ceff8d5bf3d5287b35f13ddc651 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Sun, 18 Oct 2026 20:49:25 +0100 Subject: [PATCH 1/2] Stream npm packuments and fail on oversized responses Bodies over the buffered limit returned truncated bytes, so every npm packument over 10 MiB failed to decode. Buffered reads now return ErrResponseTooLarge, and npm packuments are decoded from the body. Claude-Session: https://claude.ai/code/session_01NxsA75HBjdWQKcdx7FDTDc --- README.md | 6 + client/client.go | 149 +++++++++++++++--- client/client_test.go | 83 ++++++++-- client/errors.go | 4 + internal/npm/npm.go | 293 +++++++++++++++++++++++++----------- internal/npm/stream_test.go | 150 ++++++++++++++++++ registries.go | 8 + 7 files changed, 574 insertions(+), 119 deletions(-) create mode 100644 internal/npm/stream_test.go diff --git a/README.md b/README.md index 047185d..f75adf1 100644 --- a/README.md +++ b/README.md @@ -297,6 +297,7 @@ The default client includes: - 30 second timeout - 5 retries with exponential backoff (50ms base, 10% jitter) - Automatic retry on 429 without a usable Retry-After and on 5xx responses +- Buffered response bodies limited to 10 MiB and streamed bodies to 512 MiB; a larger body fails with `ErrResponseTooLarge` (change with `WithMaxResponseSize` and `WithMaxStreamSize`) Custom client via the top-level package: @@ -325,6 +326,11 @@ err := c.GetJSON(ctx, "https://registry.npmjs.org/lodash", &data) // Raw body body, err := c.GetBody(ctx, "https://crates.io/api/v1/crates/serde") +// Streamed JSON, for documents too large to buffer +err = c.StreamJSON(ctx, "https://registry.npmjs.org/renovate", func(dec *json.Decoder) error { + return dec.Decode(&data) +}) + // HEAD request statusCode, err := c.Head(ctx, "https://registry.npmjs.org/lodash") ``` diff --git a/client/client.go b/client/client.go index 49d3b58..157e0a6 100644 --- a/client/client.go +++ b/client/client.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "math" "math/rand" @@ -21,9 +22,11 @@ const ( backoffBase = 2 jitterFactor = 0.1 - // maxResponseSize is the maximum response body size we will read (10 MB). - // Registry JSON API responses should be well under this limit. + // maxResponseSize is the default limit for buffered response bodies. maxResponseSize = 10 << 20 + // maxStreamSize is the default limit for streamed bodies. The largest npm + // packuments are around 100 MB. + maxStreamSize = 512 << 20 ) // RateLimiter controls request pacing. @@ -38,6 +41,10 @@ type Client struct { MaxRetries int BaseDelay time.Duration RateLimiter RateLimiter + // MaxResponseSize limits buffered bodies; zero means 10 MiB. + MaxResponseSize int64 + // MaxStreamSize limits bodies read through StreamJSON; zero means 512 MiB. + MaxStreamSize int64 } // DefaultClient returns a client with sensible defaults. @@ -61,8 +68,89 @@ func (c *Client) GetJSON(ctx context.Context, url string, v any) error { return json.Unmarshal(body, v) } -// GetBody fetches a URL and returns the response body. +// GetBody fetches a URL and returns the response body. A body larger than +// the client's MaxResponseSize fails with ErrResponseTooLarge. func (c *Client) GetBody(ctx context.Context, url string) ([]byte, error) { + resp, err := c.open(ctx, url, nil) + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + return readLimited(resp.Body, url, c.maxResponseSize()) +} + +// StreamJSON fetches a URL and passes a decoder over the response body to fn, +// so large documents can be decoded without holding them in memory. Bodies +// are limited to MaxStreamSize. Retries cover getting a response; an error +// while fn reads the body is returned as is. +func (c *Client) StreamJSON(ctx context.Context, url string, fn func(*json.Decoder) error) error { + resp, err := c.open(ctx, url, nil) + if err != nil { + return err + } + defer func() { _ = resp.Body.Close() }() + limit := c.maxStreamSize() + return fn(json.NewDecoder(&limitedReader{r: resp.Body, remaining: limit, limit: limit, url: url})) +} + +func (c *Client) maxResponseSize() int64 { + if c.MaxResponseSize > 0 { + return c.MaxResponseSize + } + return maxResponseSize +} + +func (c *Client) maxStreamSize() int64 { + if c.MaxStreamSize > 0 { + return c.MaxStreamSize + } + return maxStreamSize +} + +func readLimited(r io.Reader, url string, limit int64) ([]byte, error) { + body, err := io.ReadAll(io.LimitReader(r, limit+1)) + if err != nil { + return nil, err + } + if int64(len(body)) > limit { + return nil, tooLarge(url, limit) + } + return body, nil +} + +func tooLarge(url string, limit int64) error { + return fmt.Errorf("%w: response body for %s exceeds %d bytes", ErrResponseTooLarge, url, limit) +} + +// limitedReader fails once more than limit bytes have been read, where +// io.LimitReader would end the stream early and look like a complete body. +type limitedReader struct { + r io.Reader + remaining int64 + limit int64 + url string +} + +func (l *limitedReader) Read(p []byte) (int, error) { + if l.remaining < 0 { + return 0, tooLarge(l.url, l.limit) + } + if int64(len(p)) > l.remaining+1 { + p = p[:l.remaining+1] + } + n, err := l.r.Read(p) + if int64(n) > l.remaining { + n = int(l.remaining) + l.remaining = -1 + return n, tooLarge(l.url, l.limit) + } + l.remaining -= int64(n) + return n, err +} + +// open sends the request, retrying rate limits, server errors and transport +// failures, and returns a successful response with its body unread. +func (c *Client) open(ctx context.Context, url string, header http.Header) (*http.Response, error) { var lastErr error for attempt := 0; attempt <= c.MaxRetries; attempt++ { @@ -84,9 +172,9 @@ func (c *Client) GetBody(ctx context.Context, url string) ([]byte, error) { } } - body, err := c.doRequest(ctx, url) + resp, err := c.doRequest(ctx, url, header) if err == nil { - return body, nil + return resp, nil } lastErr = err @@ -113,7 +201,7 @@ func (c *Client) GetBody(ctx context.Context, url string) ([]byte, error) { return nil, lastErr } -func (c *Client) doRequest(ctx context.Context, url string) ([]byte, error) { +func (c *Client) doRequest(ctx context.Context, url string, header http.Header) (*http.Response, error) { req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) if err != nil { return nil, err @@ -121,37 +209,38 @@ func (c *Client) doRequest(ctx context.Context, url string) ([]byte, error) { req.Header.Set("User-Agent", c.UserAgent) req.Header.Set("Accept", "application/json") + for name, values := range header { + req.Header[http.CanonicalHeaderKey(name)] = values + } resp, err := c.HTTPClient.Do(req) if err != nil { return nil, err } + if resp.StatusCode < http.StatusBadRequest { + return resp, nil + } defer func() { _ = resp.Body.Close() }() body, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseSize)) if err != nil { return nil, err } - - if resp.StatusCode >= http.StatusBadRequest { - httpErr := &HTTPError{ - StatusCode: resp.StatusCode, - URL: url, - Body: string(body), - } - if resp.StatusCode == http.StatusTooManyRequests { - if delay, ok := parseRetryAfter(resp.Header.Get("Retry-After"), time.Now()); ok { - seconds := delay / time.Second - if delay%time.Second != 0 { - seconds++ - } - return nil, &RateLimitError{RetryAfter: int(seconds)} + httpErr := &HTTPError{ + StatusCode: resp.StatusCode, + URL: url, + Body: string(body), + } + if resp.StatusCode == http.StatusTooManyRequests { + if delay, ok := parseRetryAfter(resp.Header.Get("Retry-After"), time.Now()); ok { + seconds := delay / time.Second + if delay%time.Second != 0 { + seconds++ } + return nil, &RateLimitError{RetryAfter: int(seconds)} } - return nil, httpErr } - - return body, nil + return nil, httpErr } func isHTTPError(err error, target **HTTPError) bool { @@ -220,6 +309,20 @@ func WithMaxRetries(n int) Option { } } +// WithMaxResponseSize sets the limit for buffered response bodies. +func WithMaxResponseSize(n int64) Option { + return func(c *Client) { + c.MaxResponseSize = n + } +} + +// WithMaxStreamSize sets the limit for bodies read through StreamJSON. +func WithMaxStreamSize(n int64) Option { + return func(c *Client) { + c.MaxStreamSize = n + } +} + // NewClient creates a new client with the given options. func NewClient(opts ...Option) *Client { c := DefaultClient() diff --git a/client/client_test.go b/client/client_test.go index 4e203ec..d809eab 100644 --- a/client/client_test.go +++ b/client/client_test.go @@ -2,6 +2,8 @@ package client import ( "context" + "encoding/json" + "errors" "net/http" "net/http/httptest" "strings" @@ -71,27 +73,84 @@ func TestGetJSON(t *testing.T) { } func TestResponseSizeLimited(t *testing.T) { - // Serve a response body larger than maxResponseSize (10 MB). - oversized := strings.Repeat("x", maxResponseSize+1024) - + requests := 0 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - _, _ = w.Write([]byte(oversized)) + requests++ + _, _ = w.Write([]byte(strings.Repeat("x", maxResponseSize+1))) })) defer srv.Close() c := DefaultClient() - c.MaxRetries = 0 - body, err := c.GetBody(context.Background(), srv.URL) - if err != nil { - t.Fatalf("unexpected error: %v", err) + if !errors.Is(err, ErrResponseTooLarge) { + t.Fatalf("err = %v, want ErrResponseTooLarge", err) + } + if body != nil { + t.Errorf("got %d bytes, want none", len(body)) + } + if requests != 1 { + t.Errorf("requests = %d, want no retry for an oversized body", requests) } - if len(body) > maxResponseSize { - t.Fatalf("response body should be capped at %d bytes, got %d", maxResponseSize, len(body)) + var v any + if err := c.GetJSON(context.Background(), srv.URL, &v); !errors.Is(err, ErrResponseTooLarge) { + t.Errorf("GetJSON err = %v, want ErrResponseTooLarge", err) + } +} + +func TestResponseAtLimit(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(strings.Repeat("x", 100))) + })) + defer srv.Close() + + c := NewClient(WithMaxResponseSize(100)) + if body, err := c.GetBody(context.Background(), srv.URL); err != nil || len(body) != 100 { + t.Errorf("at limit: %d bytes, %v", len(body), err) + } + c = NewClient(WithMaxResponseSize(99), WithMaxRetries(0)) + if _, err := c.GetBody(context.Background(), srv.URL); !errors.Is(err, ErrResponseTooLarge) { + t.Errorf("over a raised limit: err = %v", err) } - if len(body) != maxResponseSize { - t.Fatalf("expected exactly %d bytes from LimitReader, got %d", maxResponseSize, len(body)) +} + +func TestStreamJSON(t *testing.T) { + doc := `{"versions":{` + strings.Repeat(`"1.0.0":{"name":"x"},`, 100) + `"2.0.0":{"name":"y"}}}` + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/missing" { + http.NotFound(w, r) + return + } + _, _ = w.Write([]byte(doc)) + })) + defer srv.Close() + + c := NewClient(WithMaxResponseSize(10), WithMaxRetries(0)) + var v struct { + Versions map[string]struct{ Name string } `json:"versions"` + } + err := c.StreamJSON(context.Background(), srv.URL, func(dec *json.Decoder) error { return dec.Decode(&v) }) + if err != nil || v.Versions["2.0.0"].Name != "y" { + t.Errorf("StreamJSON = %+v, %v; the buffered limit should not apply", v, err) + } + + small := NewClient(WithMaxStreamSize(int64(len(doc)-1)), WithMaxRetries(0)) + err = small.StreamJSON(context.Background(), srv.URL, func(dec *json.Decoder) error { return dec.Decode(&v) }) + if !errors.Is(err, ErrResponseTooLarge) { + t.Errorf("over the stream limit: err = %v, want ErrResponseTooLarge", err) + } + exact := NewClient(WithMaxStreamSize(int64(len(doc))), WithMaxRetries(0)) + if err := exact.StreamJSON(context.Background(), srv.URL, func(dec *json.Decoder) error { return dec.Decode(&v) }); err != nil { + t.Errorf("at the stream limit: %v", err) + } + + var httpErr *HTTPError + err = c.StreamJSON(context.Background(), srv.URL+"/missing", func(dec *json.Decoder) error { + t.Error("fn called for a 404") + return nil + }) + if !errors.As(err, &httpErr) || !httpErr.IsNotFound() { + t.Errorf("404 err = %v", err) } } diff --git a/client/errors.go b/client/errors.go index 718bc70..3f7594c 100644 --- a/client/errors.go +++ b/client/errors.go @@ -9,6 +9,10 @@ import ( // ErrNotFound is returned when a package or version is not found. var ErrNotFound = errors.New("not found") +// ErrResponseTooLarge is returned when a response body is larger than the +// client allows. +var ErrResponseTooLarge = errors.New("response body too large") + // HTTPError represents an HTTP error response. type HTTPError struct { StatusCode int diff --git a/internal/npm/npm.go b/internal/npm/npm.go index a921161..3086aee 100644 --- a/internal/npm/npm.go +++ b/internal/npm/npm.go @@ -51,18 +51,6 @@ func (r *Registry) URLs() core.URLBuilder { //nolint:ireturn return r.urls } -type packageResponse[V any] struct { - ID string `json:"_id"` - Name string `json:"name"` - Description string `json:"description"` - Homepage interface{} `json:"homepage"` - Repository interface{} `json:"repository"` - Versions map[string]V `json:"versions"` - Time map[string]string `json:"time"` - Maintainers []maintainerInfo `json:"maintainers"` - DistTags map[string]string `json:"dist-tags"` -} - type versionInfo struct { Name string `json:"name"` Version string `json:"version"` @@ -167,46 +155,60 @@ type maintainerInfo struct { } func (r *Registry) FetchPackage(ctx context.Context, name string) (*core.Package, error) { - escapedName := url.PathEscape(name) - url := fmt.Sprintf("%s/%s", r.baseURL, escapedName) - - var resp packageResponse[json.RawMessage] - if err := r.client.GetJSON(ctx, url, &resp); err != nil { - if httpErr, ok := err.(*core.HTTPError); ok && httpErr.IsNotFound() { - return nil, &core.NotFoundError{Ecosystem: ecosystem, Name: name} + // dist-tags comes before versions in registry.npmjs.org packuments, so + // usually only the latest version is decoded. If it hasn't been seen yet, + // versions are kept raw until it is. + var latest versionInfo + var haveLatest bool + var pending map[string]json.RawMessage + var header *packumentHeader + header, err := r.streamPackument(ctx, name, func(h *packumentHeader, number string, dec *json.Decoder) error { + if want, ok := h.DistTags["latest"]; ok { + if number != want || haveLatest { + return skipValue(dec) + } + haveLatest = true + return dec.Decode(&latest) + } + var raw json.RawMessage + if err := dec.Decode(&raw); err != nil { + return err } + if pending == nil { + pending = make(map[string]json.RawMessage) + } + pending[number] = raw + return nil + }) + if err != nil { return nil, err } - latestVersion := resp.DistTags["latest"] - var latestData json.RawMessage - if latestVersion != "" { - latestData = resp.Versions[latestVersion] - } else if len(resp.Versions) > 0 { - for _, v := range resp.Versions { - latestData = v - break + latestVersion := header.DistTags["latest"] + if !haveLatest && len(pending) > 0 { + data, ok := pending[latestVersion] + if !ok { + for _, v := range pending { + data = v + break + } } - } - - var latest versionInfo - if len(latestData) > 0 { - if err := json.Unmarshal(latestData, &latest); err != nil { + if err := json.Unmarshal(data, &latest); err != nil { return nil, err } } pkg := &core.Package{ - Name: resp.ID, - Description: coalesceString(latest.Description, resp.Description), - Homepage: extractString(resp.Homepage), - Repository: core.ExtractRepoURLWithFallback(latest.Repository, resp.Repository), + Name: header.ID, + Description: coalesceString(latest.Description, header.Description), + Homepage: extractString(header.Homepage), + Repository: core.ExtractRepoURLWithFallback(latest.Repository, header.Repository), Licenses: core.ExtractLicense(latest.License), Keywords: extractKeywords(latest.Keywords), - Namespace: extractNamespace(resp.ID), + Namespace: extractNamespace(header.ID), LatestVersion: latestVersion, Metadata: map[string]any{ - "dist-tags": resp.DistTags, + "dist-tags": header.DistTags, "funding": latest.Funding, }, } @@ -214,45 +216,183 @@ func (r *Registry) FetchPackage(ctx context.Context, name string) (*core.Package return pkg, nil } -func (r *Registry) FetchVersions(ctx context.Context, name string) ([]core.Version, error) { - escapedName := url.PathEscape(name) - url := fmt.Sprintf("%s/%s", r.baseURL, escapedName) +// packumentHeader is a packument without its versions. +type packumentHeader struct { + ID string `json:"_id"` + Name string `json:"name"` + Description string `json:"description"` + Homepage interface{} `json:"homepage"` + Repository interface{} `json:"repository"` + Time map[string]string `json:"time"` + Maintainers []maintainerInfo `json:"maintainers"` + DistTags map[string]string `json:"dist-tags"` +} - var resp packageResponse[versionInfo] - if err := r.client.GetJSON(ctx, url, &resp); err != nil { +// streamPackument decodes a packument straight from the response body, so a +// document of tens of MB is never held whole. version is called with each +// entry of versions and must consume exactly one value from dec; header +// holds the top-level fields read so far. +func (r *Registry) streamPackument(ctx context.Context, name string, version func(h *packumentHeader, number string, dec *json.Decoder) error) (*packumentHeader, error) { + endpoint := fmt.Sprintf("%s/%s", r.baseURL, url.PathEscape(name)) + var header packumentHeader + err := r.client.StreamJSON(ctx, endpoint, func(dec *json.Decoder) error { + if err := expectDelim(dec, '{'); err != nil { + return err + } + for dec.More() { + tok, err := dec.Token() + if err != nil { + return err + } + key, _ := tok.(string) + var target any + switch key { + case "_id": + target = &header.ID + case "name": + target = &header.Name + case "description": + target = &header.Description + case "homepage": + target = &header.Homepage + case "repository": + target = &header.Repository + case "time": + target = &header.Time + case "maintainers": + target = &header.Maintainers + case "dist-tags": + target = &header.DistTags + case "versions": + if err := decodeVersions(dec, &header, version); err != nil { + return err + } + continue + default: + if err := skipValue(dec); err != nil { + return err + } + continue + } + if err := dec.Decode(target); err != nil { + return err + } + } + return expectDelim(dec, '}') + }) + if err != nil { if httpErr, ok := err.(*core.HTTPError); ok && httpErr.IsNotFound() { return nil, &core.NotFoundError{Ecosystem: ecosystem, Name: name} } return nil, err } + return &header, nil +} - versions := make([]core.Version, 0, len(resp.Versions)) - for num, v := range resp.Versions { - versions = append(versions, makeVersion(num, v, resp.Time[num])) +func decodeVersions(dec *json.Decoder, header *packumentHeader, version func(h *packumentHeader, number string, dec *json.Decoder) error) error { + tok, err := dec.Token() + if err != nil { + return err + } + if tok == nil { + return nil + } + if tok != json.Delim('{') { + return fmt.Errorf("npm packument: versions is not an object") + } + for dec.More() { + tok, err := dec.Token() + if err != nil { + return err + } + number, _ := tok.(string) + if err := version(header, number, dec); err != nil { + return err + } } + return expectDelim(dec, '}') +} - return versions, nil +func expectDelim(dec *json.Decoder, want json.Delim) error { + tok, err := dec.Token() + if err != nil { + return err + } + if tok != want { + return fmt.Errorf("npm packument: expected %q, got %v", want, tok) + } + return nil } -// FetchVersion retrieves one release while retaining packument publication times. -func (r *Registry) FetchVersion(ctx context.Context, name, version string) (*core.Version, error) { - endpoint := fmt.Sprintf("%s/%s", r.baseURL, url.PathEscape(name)) - var resp packageResponse[json.RawMessage] - if err := r.client.GetJSON(ctx, endpoint, &resp); err != nil { - if httpErr, ok := err.(*core.HTTPError); ok && httpErr.IsNotFound() { - return nil, &core.NotFoundError{Ecosystem: ecosystem, Name: name} +// skipValue consumes one value without keeping it. +func skipValue(dec *json.Decoder) error { + depth := 0 + for { + tok, err := dec.Token() + if err != nil { + return err + } + switch tok { + case json.Delim('{'), json.Delim('['): + depth++ + case json.Delim('}'), json.Delim(']'): + depth-- } + if depth == 0 { + return nil + } + } +} + +// decodeOne decodes the version named want and skips the rest. +func decodeOne(want string, info *versionInfo, found *bool) func(*packumentHeader, string, *json.Decoder) error { + return func(_ *packumentHeader, number string, dec *json.Decoder) error { + if number != want || *found { + return skipValue(dec) + } + *found = true + return dec.Decode(info) + } +} + +func (r *Registry) FetchVersions(ctx context.Context, name string) ([]core.Version, error) { + var versions []core.Version + header, err := r.streamPackument(ctx, name, func(_ *packumentHeader, number string, dec *json.Decoder) error { + var v versionInfo + if err := dec.Decode(&v); err != nil { + return err + } + versions = append(versions, makeVersion(number, v, "")) + return nil + }) + if err != nil { return nil, err } - data, ok := resp.Versions[version] - if !ok { - return nil, &core.NotFoundError{Ecosystem: ecosystem, Name: name, Version: version} + + // time can come after versions, so publication times are filled in last. + for i := range versions { + if ts := header.Time[versions[i].Number]; ts != "" { + versions[i].PublishedAt, _ = time.Parse(time.RFC3339, ts) + } + } + if versions == nil { + versions = []core.Version{} } + return versions, nil +} + +// FetchVersion retrieves one release while retaining packument publication times. +func (r *Registry) FetchVersion(ctx context.Context, name, version string) (*core.Version, error) { var info versionInfo - if err := json.Unmarshal(data, &info); err != nil { + var found bool + header, err := r.streamPackument(ctx, name, decodeOne(version, &info, &found)) + if err != nil { return nil, err } - result := makeVersion(version, info, resp.Time[version]) + if !found { + return nil, &core.NotFoundError{Ecosystem: ecosystem, Name: name, Version: version} + } + result := makeVersion(version, info, header.Time[version]) return &result, nil } @@ -292,25 +432,14 @@ func makeVersion(num string, v versionInfo, timeStr string) core.Version { } func (r *Registry) FetchDependencies(ctx context.Context, name, version string) ([]core.Dependency, error) { - escapedName := url.PathEscape(name) - url := fmt.Sprintf("%s/%s", r.baseURL, escapedName) - - var resp packageResponse[json.RawMessage] - if err := r.client.GetJSON(ctx, url, &resp); err != nil { - if httpErr, ok := err.(*core.HTTPError); ok && httpErr.IsNotFound() { - return nil, &core.NotFoundError{Ecosystem: ecosystem, Name: name} - } + var v versionInfo + var found bool + if _, err := r.streamPackument(ctx, name, decodeOne(version, &v, &found)); err != nil { return nil, err } - - data, ok := resp.Versions[version] - if !ok { + if !found { return nil, &core.NotFoundError{Ecosystem: ecosystem, Name: name, Version: version} } - var v versionInfo - if err := json.Unmarshal(data, &v); err != nil { - return nil, err - } var deps []core.Dependency @@ -343,19 +472,15 @@ func (r *Registry) FetchDependencies(ctx context.Context, name, version string) } func (r *Registry) FetchMaintainers(ctx context.Context, name string) ([]core.Maintainer, error) { - escapedName := url.PathEscape(name) - url := fmt.Sprintf("%s/%s", r.baseURL, escapedName) - - var resp packageResponse[json.RawMessage] - if err := r.client.GetJSON(ctx, url, &resp); err != nil { - if httpErr, ok := err.(*core.HTTPError); ok && httpErr.IsNotFound() { - return nil, &core.NotFoundError{Ecosystem: ecosystem, Name: name} - } + header, err := r.streamPackument(ctx, name, func(_ *packumentHeader, _ string, dec *json.Decoder) error { + return skipValue(dec) + }) + if err != nil { return nil, err } - maintainers := make([]core.Maintainer, len(resp.Maintainers)) - for i, m := range resp.Maintainers { + maintainers := make([]core.Maintainer, len(header.Maintainers)) + for i, m := range header.Maintainers { maintainers[i] = core.Maintainer{ UUID: m.Name, Login: m.Name, diff --git a/internal/npm/stream_test.go b/internal/npm/stream_test.go new file mode 100644 index 0000000..3e51505 --- /dev/null +++ b/internal/npm/stream_test.go @@ -0,0 +1,150 @@ +package npm + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/git-pkgs/registries/client" + "github.com/git-pkgs/registries/internal/core" +) + +// largePackument builds a packument in registry.npmjs.org key order with +// enough versions to pass the client's 10 MiB buffered limit. A long readme +// sits between versions and time, as in real packuments. +func largePackument(versions int) string { + var b strings.Builder + b.WriteString(`{"_id":"big","_rev":"1-abc","name":"big","dist-tags":{"latest":"1.0.`) + fmt.Fprintf(&b, "%d", versions-1) + b.WriteString(`"},"versions":{`) + padding := strings.Repeat("x", 2000) + for i := range versions { + if i > 0 { + b.WriteByte(',') + } + fmt.Fprintf(&b, `"1.0.%d":{"name":"big","version":"1.0.%d","description":"%s","license":"MIT",`+ + `"dependencies":{"dep":"^%d.0.0"},"dist":{"integrity":"sha512-v%d","tarball":"https://registry.npmjs.org/big/-/big-1.0.%d.tgz"}}`, + i, i, padding, i, i, i) + } + b.WriteString(`},"readme":"` + strings.Repeat("r", 1<<20) + `","time":{"created":"2020-01-01T00:00:00.000Z"`) + for i := range versions { + fmt.Fprintf(&b, `,"1.0.%d":"2021-01-01T00:00:%02d.000Z"`, i, i%60) + } + b.WriteString(`},"maintainers":[{"name":"mona","email":"mona@example.com"}]}`) + return b.String() +} + +func TestLargePackument(t *testing.T) { + doc := largePackument(6000) + if len(doc) <= 10<<20 { + t.Fatalf("fixture is %d bytes, want more than the 10 MiB buffered limit", len(doc)) + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/big" { + http.NotFound(w, r) + return + } + _, _ = w.Write([]byte(doc)) + })) + defer server.Close() + reg := New(server.URL, core.DefaultClient()) + ctx := context.Background() + + versions, err := reg.FetchVersions(ctx, "big") + if err != nil { + t.Fatalf("FetchVersions failed: %v", err) + } + if len(versions) != 6000 { + t.Fatalf("got %d versions, want 6000", len(versions)) + } + v := versions[1234] + if v.Number != "1.0.1234" || v.Integrity != "sha512-v1234" || v.Licenses != "MIT" { + t.Errorf("version = %+v", v) + } + if want := time.Date(2021, 1, 1, 0, 0, 34, 0, time.UTC); !v.PublishedAt.Equal(want) { + t.Errorf("PublishedAt = %v, want %v from time after versions", v.PublishedAt, want) + } + + pkg, err := reg.FetchPackage(ctx, "big") + if err != nil || pkg.LatestVersion != "1.0.5999" || pkg.Licenses != "MIT" || pkg.Name != "big" { + t.Errorf("FetchPackage = %+v, %v", pkg, err) + } + + one, err := reg.FetchVersion(ctx, "big", "1.0.42") + if err != nil || one.Integrity != "sha512-v42" || one.PublishedAt.Second() != 42 { + t.Errorf("FetchVersion = %+v, %v", one, err) + } + + deps, err := reg.FetchDependencies(ctx, "big", "1.0.7") + if err != nil || len(deps) != 1 || deps[0].Requirements != "^7.0.0" { + t.Errorf("FetchDependencies = %+v, %v", deps, err) + } + + maintainers, err := reg.FetchMaintainers(ctx, "big") + if err != nil || len(maintainers) != 1 || maintainers[0].Login != "mona" { + t.Errorf("FetchMaintainers = %+v, %v", maintainers, err) + } + + var notFound *core.NotFoundError + if _, err := reg.FetchVersion(ctx, "big", "9.9.9"); !errors.As(err, ¬Found) || notFound.Version != "9.9.9" { + t.Errorf("missing version err = %v", err) + } + if _, err := reg.FetchDependencies(ctx, "big", "9.9.9"); !errors.As(err, ¬Found) { + t.Errorf("missing version deps err = %v", err) + } + if _, err := reg.FetchVersions(ctx, "missing"); !errors.As(err, ¬Found) { + t.Errorf("missing package err = %v", err) + } +} + +func TestFetchPackageWithDistTagsAfterVersions(t *testing.T) { + doc := `{"_id":"late","versions":{"1.0.0":{"description":"old"},"2.0.0":{"description":"new"}},"dist-tags":{"latest":"2.0.0"}}` + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(doc)) + })) + defer server.Close() + + pkg, err := New(server.URL, core.DefaultClient()).FetchPackage(context.Background(), "late") + if err != nil || pkg.LatestVersion != "2.0.0" || pkg.Description != "new" { + t.Errorf("FetchPackage = %+v, %v", pkg, err) + } +} + +func TestStreamPackumentErrors(t *testing.T) { + tests := map[string]string{ + "truncated": `{"_id":"x","versions":{"1.0.0":{"name":"x"`, + "versions not an object": `{"_id":"x","versions":[1,2]}`, + "not an object": `[]`, + "trailing garbage": `{"_id":"x"`, + } + for name, doc := range tests { + t.Run(name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(doc)) + })) + defer server.Close() + if _, err := New(server.URL, core.DefaultClient()).FetchVersions(context.Background(), "x"); err == nil { + t.Error("expected an error") + } + }) + } +} + +func TestPackumentOverStreamLimit(t *testing.T) { + doc := largePackument(50) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(doc)) + })) + defer server.Close() + + c := client.NewClient(client.WithMaxStreamSize(int64(len(doc)/2)), client.WithMaxRetries(0)) + _, err := New(server.URL, c).FetchVersions(context.Background(), "big") + if !errors.Is(err, client.ErrResponseTooLarge) { + t.Errorf("err = %v, want ErrResponseTooLarge", err) + } +} diff --git a/registries.go b/registries.go index 13250f0..3311dd7 100644 --- a/registries.go +++ b/registries.go @@ -97,6 +97,8 @@ const ( // Re-export errors var ( ErrNotFound = client.ErrNotFound + // ErrResponseTooLarge is returned when a response body exceeds the client limit. + ErrResponseTooLarge = client.ErrResponseTooLarge ) // Error types @@ -146,6 +148,12 @@ var WithTransport = client.WithTransport // WithSafeHTTP applies protections for requests to untrusted hosts. var WithSafeHTTP = client.WithSafeHTTP +// WithMaxResponseSize sets the limit for buffered response bodies. +var WithMaxResponseSize = client.WithMaxResponseSize + +// WithMaxStreamSize sets the limit for streamed response bodies. +var WithMaxStreamSize = client.WithMaxStreamSize + // SupportedEcosystems returns all registered ecosystem types. // Note: ecosystems must be imported to be registered. func SupportedEcosystems() []string { From 053b699c332b758b94bd75feb8dd5f99b5c095e6 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Mon, 19 Oct 2026 08:22:50 +0100 Subject: [PATCH 2/2] Preserve buffered retries and validate streamed JSON responses --- client/client.go | 42 ++++++++++++++------ client/client_test.go | 79 +++++++++++++++++++++++++++++++++++++ internal/npm/stream_test.go | 17 +++++++- 3 files changed, 124 insertions(+), 14 deletions(-) diff --git a/client/client.go b/client/client.go index 157e0a6..25e8fbc 100644 --- a/client/client.go +++ b/client/client.go @@ -71,26 +71,39 @@ func (c *Client) GetJSON(ctx context.Context, url string, v any) error { // GetBody fetches a URL and returns the response body. A body larger than // the client's MaxResponseSize fails with ErrResponseTooLarge. func (c *Client) GetBody(ctx context.Context, url string) ([]byte, error) { - resp, err := c.open(ctx, url, nil) - if err != nil { - return nil, err - } - defer func() { _ = resp.Body.Close() }() - return readLimited(resp.Body, url, c.maxResponseSize()) + var body []byte + _, err := c.open(ctx, url, nil, func(resp *http.Response) error { + defer func() { _ = resp.Body.Close() }() + var err error + body, err = readLimited(resp.Body, url, c.maxResponseSize()) + return err + }) + return body, err } // StreamJSON fetches a URL and passes a decoder over the response body to fn, // so large documents can be decoded without holding them in memory. Bodies // are limited to MaxStreamSize. Retries cover getting a response; an error -// while fn reads the body is returned as is. +// while fn reads the body is returned as is. fn must consume the complete +// JSON document; trailing data and body read errors are rejected. func (c *Client) StreamJSON(ctx context.Context, url string, fn func(*json.Decoder) error) error { - resp, err := c.open(ctx, url, nil) + resp, err := c.open(ctx, url, nil, nil) if err != nil { return err } defer func() { _ = resp.Body.Close() }() limit := c.maxStreamSize() - return fn(json.NewDecoder(&limitedReader{r: resp.Body, remaining: limit, limit: limit, url: url})) + dec := json.NewDecoder(&limitedReader{r: resp.Body, remaining: limit, limit: limit, url: url}) + if err := fn(dec); err != nil { + return err + } + if _, err := dec.Token(); err != io.EOF { + if err != nil { + return err + } + return errors.New("unexpected data after JSON document") + } + return nil } func (c *Client) maxResponseSize() int64 { @@ -148,9 +161,9 @@ func (l *limitedReader) Read(p []byte) (int, error) { return n, err } -// open sends the request, retrying rate limits, server errors and transport -// failures, and returns a successful response with its body unread. -func (c *Client) open(ctx context.Context, url string, header http.Header) (*http.Response, error) { +// open retries requests and optional body reads within one retry budget. +// read must close the body; a nil read leaves it open for the caller. +func (c *Client) open(ctx context.Context, url string, header http.Header, read func(*http.Response) error) (*http.Response, error) { var lastErr error for attempt := 0; attempt <= c.MaxRetries; attempt++ { @@ -173,12 +186,15 @@ func (c *Client) open(ctx context.Context, url string, header http.Header) (*htt } resp, err := c.doRequest(ctx, url, header) + if err == nil && read != nil { + err = read(resp) + } if err == nil { return resp, nil } lastErr = err - if errors.Is(err, ErrUpstreamHeld) { + if errors.Is(err, ErrUpstreamHeld) || errors.Is(err, ErrResponseTooLarge) { return nil, err } var rateErr *RateLimitError diff --git a/client/client_test.go b/client/client_test.go index d809eab..c28cd7c 100644 --- a/client/client_test.go +++ b/client/client_test.go @@ -4,8 +4,10 @@ import ( "context" "encoding/json" "errors" + "io" "net/http" "net/http/httptest" + "strconv" "strings" "testing" ) @@ -72,6 +74,83 @@ func TestGetJSON(t *testing.T) { } } +func TestGetBodyRetriesTruncatedResponse(t *testing.T) { + for _, failures := range []int{1, 3} { + t.Run(strconv.Itoa(failures), func(t *testing.T) { + requests := 0 + payload := `{"ok":true}` + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + requests++ + w.Header().Set("Content-Length", strconv.Itoa(len(payload))) + if requests <= failures { + _, _ = w.Write([]byte(payload[:2])) + return + } + _, _ = w.Write([]byte(payload)) + })) + defer srv.Close() + + c := NewClient(WithMaxRetries(1)) + c.BaseDelay = 0 + body, err := c.GetBody(context.Background(), srv.URL) + if failures == 1 { + if err != nil || string(body) != payload { + t.Errorf("GetBody = %q, %v", body, err) + } + } else if !errors.Is(err, io.ErrUnexpectedEOF) || body != nil { + t.Errorf("GetBody = %q, %v; want no body and unexpected EOF", body, err) + } + if requests != 2 { + t.Errorf("requests = %d, want 2", requests) + } + }) + } +} + +func TestStreamJSONValidatesEndOfResponse(t *testing.T) { + for _, tt := range []struct { + name string + suffix string + wantErr bool + tooLarge bool + truncated bool + }{ + {name: "whitespace", suffix: " \n\t"}, + {name: "garbage", suffix: "invalid", wantErr: true}, + {name: "second value", suffix: "{}", wantErr: true}, + {name: "oversized whitespace", suffix: strings.Repeat(" ", 1024), wantErr: true, tooLarge: true}, + {name: "truncated transfer", wantErr: true, truncated: true}, + } { + t.Run(tt.name, func(t *testing.T) { + requests := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + requests++ + doc := `{"ok":true}` + tt.suffix + if tt.truncated { + w.Header().Set("Content-Length", strconv.Itoa(len(doc)+1)) + } + _, _ = w.Write([]byte(doc)) + })) + defer srv.Close() + + c := NewClient(WithMaxStreamSize(64)) + err := c.StreamJSON(context.Background(), srv.URL, func(dec *json.Decoder) error { + var value map[string]bool + return dec.Decode(&value) + }) + if (err != nil) != tt.wantErr || errors.Is(err, ErrResponseTooLarge) != tt.tooLarge { + t.Errorf("StreamJSON error = %v", err) + } + if tt.truncated && !errors.Is(err, io.ErrUnexpectedEOF) { + t.Errorf("StreamJSON error = %v, want unexpected EOF", err) + } + if requests != 1 { + t.Errorf("requests = %d, want 1", requests) + } + }) + } +} + func TestResponseSizeLimited(t *testing.T) { requests := 0 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/internal/npm/stream_test.go b/internal/npm/stream_test.go index 3e51505..333a3b9 100644 --- a/internal/npm/stream_test.go +++ b/internal/npm/stream_test.go @@ -120,7 +120,8 @@ func TestStreamPackumentErrors(t *testing.T) { "truncated": `{"_id":"x","versions":{"1.0.0":{"name":"x"`, "versions not an object": `{"_id":"x","versions":[1,2]}`, "not an object": `[]`, - "trailing garbage": `{"_id":"x"`, + "trailing garbage": `{"_id":"x"}invalid`, + "second value": `{"_id":"x"}{}`, } for name, doc := range tests { t.Run(name, func(t *testing.T) { @@ -148,3 +149,17 @@ func TestPackumentOverStreamLimit(t *testing.T) { t.Errorf("err = %v, want ErrResponseTooLarge", err) } } + +func TestPackumentWhitespaceOverStreamLimit(t *testing.T) { + doc := `{"_id":"x","versions":{}}` + strings.Repeat(" ", 1024) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(doc)) + })) + defer server.Close() + + c := client.NewClient(client.WithMaxStreamSize(64)) + _, err := New(server.URL, c).FetchPackage(context.Background(), "x") + if !errors.Is(err, client.ErrResponseTooLarge) { + t.Errorf("err = %v, want ErrResponseTooLarge", err) + } +}