diff --git a/cmd/proxy/main.go b/cmd/proxy/main.go index 258fba32..5c338806 100644 --- a/cmd/proxy/main.go +++ b/cmd/proxy/main.go @@ -554,6 +554,7 @@ func runMirror() { proxy.CacheMetadata = true // mirror always caches metadata proxy.MetadataTTL = cfg.ParseMetadataTTL() proxy.MetadataMaxSize = cfg.ParseMetadataMaxSize() + proxy.MetadataStreamMaxSize = cfg.ParseMetadataStreamMaxSize() // As serve does, resolve Swift source repositories in the denylist through // the upstream, and refuse to run when one can't be. if err := server.ConfigureDenylist(cfg, proxy); err != nil { diff --git a/docs/architecture.md b/docs/architecture.md index 24af2fbb..4fcb798f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -316,7 +316,7 @@ HTTP protocol handlers for each registry type. ### `internal/metadata/npm` -The npm metadata rewrite, separated from the handler so it can run where the proxy's infrastructure can't, such as WASI. It takes a packument's bytes and everything else as explicit inputs (proxy URL, cooldown policy, the time cooldown is evaluated at, denied versions, pushed versions) and returns the rewritten document with the versions it kept. It filters versions by cooldown and the denylist, moves `latest` off a withheld version, lists pushed versions in place of upstream ones, and points each tarball URL at the proxy, copying everything else byte for byte. It does no I/O and never reads the clock; callbacks report what it did, which the handler logs. It imports no database, storage, metrics, HTTP or logging packages, which a test checks. +The npm metadata rewrite, separated from the handler so it can run where the proxy's infrastructure can't, such as WASI. It takes a packument's bytes and everything else as explicit inputs (proxy URL, cooldown policy, the time cooldown is evaluated at, denied versions, pushed versions) and returns the rewritten document with the versions it kept. It filters versions by cooldown and the denylist, moves `latest` off a withheld version, lists pushed versions in place of upstream ones, and points each tarball URL at the proxy, copying everything else byte for byte. It does no I/O and never reads the clock; callbacks report what it did, which the handler logs. `ScanPackument` and `RewriteStream` do the same rewrite over a reader in two passes, the first collecting the time map, dist-tags and version names and the second copying each kept version, so the handler can rewrite a packument from storage without holding it in memory. It imports no database, storage, metrics, HTTP or logging packages, which a test checks. ### `internal/jsonscan` diff --git a/docs/configuration.md b/docs/configuration.md index 3e94531a..34de0d22 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1411,13 +1411,14 @@ When upstream is unreachable and the cached entry is past its TTL, the proxy ser ### Metadata size limit -Upstream metadata responses are buffered in memory before being rewritten and served. `metadata_max_size` caps that buffer to protect against OOM from a misbehaving upstream. Some npm packages with thousands of versions (for example `renovate`) exceed the 100 MB default, so raise this if you see `metadata response exceeds size limit` in the logs. Conda `repodata.json` with cooldown enabled is filtered as it streams and isn't held in memory decompressed. With `cache_metadata` the limit applies to the compressed document as it does without cooldown; one over it is filtered as it streams from the upstream and not cached. +Upstream metadata responses are buffered in memory before being rewritten and served. `metadata_max_size` caps that buffer to protect against OOM from a misbehaving upstream. Some npm packages with thousands of versions (for example `renovate`) exceed the 100 MB default, so raise this if you see `metadata response exceeds size limit` in the logs. Conda `repodata.json` with cooldown enabled is filtered as it streams and isn't held in memory decompressed. With `cache_metadata` the limit applies to the compressed document as it does without cooldown; one over it is filtered as it streams from the upstream and not cached. With `cache_metadata` and the metadata rewrite cache on, an npm packument is streamed from the upstream into storage and rewritten from there in two passes, so it isn't held in memory and the limit doesn't apply to it; the rewritten copy is kept gzip-compressed. Requests for a single version (`//`) and tarball downloads read the version from the stored copy the same way. `metadata_stream_max_size` (1 GB by default) caps a document streamed this way; one over it isn't stored, and the stored copy, if any, is served instead. ```yaml -metadata_max_size: "100MB" # default +metadata_max_size: "100MB" # default +metadata_stream_max_size: "1GB" # default ``` -Or via environment variable: `PROXY_METADATA_MAX_SIZE=250MB`. +Or via environment variable: `PROXY_METADATA_MAX_SIZE=250MB`, and `PROXY_METADATA_STREAM_MAX_SIZE=2GB` for `metadata_stream_max_size`. ### Artifact size limit diff --git a/internal/config/config.go b/internal/config/config.go index c09a1bae..25fd63db 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -170,6 +170,11 @@ type Config struct { // size return ErrMetadataTooLarge. Default: "100MB". MetadataMaxSize string `json:"metadata_max_size" yaml:"metadata_max_size"` + // MetadataStreamMaxSize is the largest metadata document the proxy + // streams from upstream into storage instead of buffering it, which + // metadata_max_size doesn't cover (e.g. "1GB"). Default: "1GB". + MetadataStreamMaxSize string `json:"metadata_stream_max_size" yaml:"metadata_stream_max_size"` + // ArtifactMaxSize is the largest upstream artifact the proxy will fetch // (e.g. "2GB"). Larger artifacts are refused and not cached. Default: "" // (unlimited). "0" is also unlimited. @@ -1481,6 +1486,7 @@ func (c *Config) LoadFromEnv() { setEnvStringSlice(&c.Hosted.ComposerVendors, "PROXY_HOSTED_COMPOSER_VENDORS") setEnvString(&c.MetadataTTL, "PROXY_METADATA_TTL") setEnvString(&c.MetadataMaxSize, "PROXY_METADATA_MAX_SIZE") + setEnvString(&c.MetadataStreamMaxSize, "PROXY_METADATA_STREAM_MAX_SIZE") setEnvString(&c.ArtifactMaxSize, "PROXY_ARTIFACT_MAX_SIZE") setEnvString(&c.MetadataRewriteCacheSize, "PROXY_METADATA_REWRITE_CACHE_SIZE") setEnvString(&c.HTTPTimeout, "PROXY_HTTP_TIMEOUT") @@ -1727,6 +1733,7 @@ const ( defaultHTTPTimeout = 30 * time.Second //nolint:mnd // sensible default defaultHitFlushInterval = time.Second defaultMetadataMaxSize = 100 << 20 + defaultMetadataStreamMaxSize = 1 << 30 defaultMetadataRewriteCacheSize = 256 << 20 defaultGradleBuildCacheMaxUploadSize = 100 << 20 defaultGradleBuildCacheSweepInterval = 10 * time.Minute @@ -1780,22 +1787,25 @@ func validateRequestTimeout(field, s string) error { return nil } -func validateMetadataMaxSize(s string) error { +func validateMetadataMaxSize(name, s string) error { if s == "" { return nil } size, err := ParseSize(s) if err != nil { - return fmt.Errorf("invalid metadata_max_size: %w", err) + return fmt.Errorf("invalid %s: %w", name, err) } if size <= 0 { - return fmt.Errorf("invalid metadata_max_size %q: must be positive", s) + return fmt.Errorf("invalid %s %q: must be positive", name, s) } return nil } func (c *Config) validateDownloadSizeLimits() error { - if err := validateMetadataMaxSize(c.MetadataMaxSize); err != nil { + if err := validateMetadataMaxSize("metadata_max_size", c.MetadataMaxSize); err != nil { + return err + } + if err := validateMetadataMaxSize("metadata_stream_max_size", c.MetadataStreamMaxSize); err != nil { return err } return validateArtifactMaxSize(c.ArtifactMaxSize) @@ -1858,6 +1868,19 @@ func (c *Config) ParseMetadataMaxSize() int64 { return size } +// ParseMetadataStreamMaxSize returns the size limit for metadata streamed +// into storage in bytes. Returns 1GB if unset or invalid. +func (c *Config) ParseMetadataStreamMaxSize() int64 { + if c.MetadataStreamMaxSize == "" { + return defaultMetadataStreamMaxSize + } + size, err := ParseSize(c.MetadataStreamMaxSize) + if err != nil || size <= 0 { + return defaultMetadataStreamMaxSize + } + return size +} + // ParseArtifactMaxSize returns the artifact size limit in bytes, or 0 for // no limit. func (c *Config) ParseArtifactMaxSize() int64 { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index a0132cfd..00f45d95 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -785,6 +785,29 @@ func TestParseMetadataMaxSize(t *testing.T) { } } +func TestParseMetadataStreamMaxSize(t *testing.T) { + cfg := Default() + if got := cfg.ParseMetadataStreamMaxSize(); got != defaultMetadataStreamMaxSize { + t.Errorf("unset: %d, want %d", got, defaultMetadataStreamMaxSize) + } + cfg.MetadataStreamMaxSize = "2GB" + if got := cfg.ParseMetadataStreamMaxSize(); got != 2<<30 { + t.Errorf("2GB: %d", got) + } + t.Setenv("PROXY_METADATA_STREAM_MAX_SIZE", "300MB") + cfg = Default() + cfg.LoadFromEnv() + if got := cfg.ParseMetadataStreamMaxSize(); got != 300<<20 { + t.Errorf("from env: %d", got) + } + for _, bad := range []string{"0", "lots"} { + cfg.MetadataStreamMaxSize = bad + if err := cfg.Validate(); err == nil { + t.Errorf("metadata_stream_max_size %q validated", bad) + } + } +} + func TestValidateMetadataMaxSize(t *testing.T) { cfg := Default() cfg.MetadataMaxSize = "not-a-size" diff --git a/internal/handler/conda.go b/internal/handler/conda.go index 3cbaf937..93ef0f02 100644 --- a/internal/handler/conda.go +++ b/internal/handler/conda.go @@ -409,7 +409,8 @@ func (h *CondaHandler) rememberRepodata(cacheKey, digest string, filter *repodat h.repodataStates[cacheKey] = repodataState{digest: digest, etag: etag, until: filter.until} } -// countingWriter records whether anything has reached the response yet. +// countingWriter counts the bytes written through it, such as whether +// anything has reached the response yet. type countingWriter struct { w io.Writer n int64 diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 4b7ebe72..bc9256ae 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -142,6 +142,10 @@ func ifNoneMatchHits(header, etag string) bool { // defaultMetadataMaxSize is used when Proxy.MetadataMaxSize is unset. const defaultMetadataMaxSize = 100 << 20 +// defaultMetadataStreamMaxSize is used when Proxy.MetadataStreamMaxSize is +// unset. +const defaultMetadataStreamMaxSize = 1 << 30 + // ErrMetadataTooLarge is returned when upstream metadata exceeds the configured limit. var ErrMetadataTooLarge = errors.New("metadata response exceeds size limit") @@ -215,14 +219,17 @@ type Proxy struct { Cooldown CooldownPolicy // ShortestCooldown is the shortest cooldown configured anywhere. It // bounds how long a rewrite filtered by cooldown is cached. - ShortestCooldown time.Duration - Denylist *denylist.Policy - CacheMetadata bool - MetadataTTL time.Duration - MetadataMaxSize int64 - GradleReadOnly bool - HostedEnabled bool - GradleMaxUploadSize int64 + ShortestCooldown time.Duration + Denylist *denylist.Policy + CacheMetadata bool + MetadataTTL time.Duration + MetadataMaxSize int64 + // MetadataStreamMaxSize caps a metadata document streamed from upstream + // into storage, which metadata_max_size doesn't cover. Default 1 GB. + MetadataStreamMaxSize int64 + GradleReadOnly bool + HostedEnabled bool + GradleMaxUploadSize int64 // NPMFullMetadata requests full npm packuments from upstream even when // cooldown is disabled, so served metadata carries publish times. NPMFullMetadata bool @@ -1816,6 +1823,10 @@ type metadataResult struct { // fromCache is set when the body is a stored copy within its TTL, read // without asking upstream. fromCache bool + // entry is the cache row of the stored document, for a lookup that + // leaves the document in storage (see storedMetadataDocument). body is + // nil then. + entry *database.MetadataCacheEntry } // describedBy returns res with the digest that entry, the cache row its body @@ -1955,24 +1966,9 @@ type upstreamMetadata struct { // alongside for the caller to store and replay. An empty acceptEncoding leaves Go to negotiate and // decompress transparently. func (p *Proxy) fetchUpstreamMetadata(ctx context.Context, upstreamURL string, entry *database.MetadataCacheEntry, accept, acceptEncoding string) (*upstreamMetadata, error) { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, upstreamURL, nil) - if err != nil { - return nil, fmt.Errorf("creating request: %w", err) - } - req.Header.Set("Accept", accept) - if acceptEncoding != "" { - req.Header.Set(headerAcceptEncoding, acceptEncoding) - } - - if entry != nil && entry.ETag.Valid { - req.Header.Set("If-None-Match", entry.ETag.String) - } else if entry != nil && entry.LastModified.Valid { - req.Header.Set("If-Modified-Since", entry.LastModified.Time.UTC().Format(http.TimeFormat)) - } - - resp, err := p.HTTPClient.Do(req) + resp, err := p.requestUpstreamMetadata(ctx, upstreamURL, entry, accept, acceptEncoding) if err != nil { - return nil, fmt.Errorf("fetching metadata: %w", err) + return nil, err } defer func() { _ = resp.Body.Close() }() @@ -2000,12 +1996,8 @@ func (p *Proxy) fetchUpstreamMetadata(ctx context.Context, upstreamURL string, e return meta, nil } - if resp.StatusCode == http.StatusNotFound { - body, _ := io.ReadAll(io.LimitReader(resp.Body, upstreamNotFoundBodyLimit)) - return nil, &upstreamNotFoundError{body: body} - } - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("upstream returned %d", resp.StatusCode) + if err := upstreamMetadataStatus(resp); err != nil { + return nil, err } // ContentLength is -1 when the transport decompresses the body, and @@ -2030,6 +2022,44 @@ func (p *Proxy) fetchUpstreamMetadata(ctx context.Context, upstreamURL string, e return meta, nil } +// requestUpstreamMetadata sends the request for a metadata document, +// revalidating entry when it is given. +func (p *Proxy) requestUpstreamMetadata(ctx context.Context, upstreamURL string, entry *database.MetadataCacheEntry, accept, acceptEncoding string) (*http.Response, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, upstreamURL, nil) + if err != nil { + return nil, fmt.Errorf("creating request: %w", err) + } + req.Header.Set("Accept", accept) + if acceptEncoding != "" { + req.Header.Set(headerAcceptEncoding, acceptEncoding) + } + + if entry != nil && entry.ETag.Valid { + req.Header.Set("If-None-Match", entry.ETag.String) + } else if entry != nil && entry.LastModified.Valid { + req.Header.Set("If-Modified-Since", entry.LastModified.Time.UTC().Format(http.TimeFormat)) + } + + resp, err := p.HTTPClient.Do(req) + if err != nil { + return nil, fmt.Errorf("fetching metadata: %w", err) + } + return resp, nil +} + +// upstreamMetadataStatus returns the error for a metadata response other +// than 200. Callers handle a 304 they asked for before. +func upstreamMetadataStatus(resp *http.Response) error { + switch resp.StatusCode { + case http.StatusOK: + return nil + case http.StatusNotFound: + body, _ := io.ReadAll(io.LimitReader(resp.Body, upstreamNotFoundBodyLimit)) + return &upstreamNotFoundError{body: body} + } + return fmt.Errorf("upstream returned %d", resp.StatusCode) +} + // refreshMetadataBlob records that upstream answered 304 for the cached // document meta was read from, by moving the row's fetched_at on. The bytes in // storage are unchanged, so they are neither written nor hashed again. It @@ -2057,14 +2087,22 @@ func (p *Proxy) refreshMetadataBlob(ecosystem, cacheKey string, meta *upstreamMe // database. It returns the digest it recorded for the bytes, or "" when they // were not cached. func (p *Proxy) cacheMetadataBlob(ctx context.Context, ecosystem, cacheKey, storagePath string, meta *upstreamMetadata) string { + digest, _ := p.storeMetadataBlob(ctx, ecosystem, cacheKey, storagePath, bytes.NewReader(meta.body), meta) + return digest +} + +// storeMetadataBlob is cacheMetadataBlob for a document read from body, +// which meta describes. It also returns the error that kept it from caching +// the document. +func (p *Proxy) storeMetadataBlob(ctx context.Context, ecosystem, cacheKey, storagePath string, body io.Reader, meta *upstreamMetadata) (string, error) { if p.DB == nil || p.Storage == nil { - return "" + return "", errMetadataNotStored } - size, hash, err := p.Storage.Store(ctx, storagePath, bytes.NewReader(meta.body)) + size, hash, err := p.Storage.Store(ctx, storagePath, body) if err != nil { p.Logger.Warn("failed to cache metadata", "ecosystem", ecosystem, "key", cacheKey, "error", err) - return "" + return "", err } var digest string if hash != "" { @@ -2094,9 +2132,9 @@ func (p *Proxy) cacheMetadataBlob(ctx context.Context, ecosystem, cacheKey, stor if delErr := p.Storage.Delete(ctx, storagePath); delErr != nil { p.Logger.Warn("failed to discard metadata blob", "ecosystem", ecosystem, "key", cacheKey, "error", delErr) } - return "" + return "", err } - return digest + return digest, nil } // currentMetadataEntry re-reads the metadata cache row and returns it, or diff --git a/internal/handler/metadata_stored.go b/internal/handler/metadata_stored.go new file mode 100644 index 00000000..ffe524d9 --- /dev/null +++ b/internal/handler/metadata_stored.go @@ -0,0 +1,317 @@ +package handler + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "hash" + "io" + "net/http" + "time" + + "github.com/git-pkgs/proxy/internal/database" + "github.com/git-pkgs/proxy/internal/metrics" +) + +// A metadata document can be rewritten from storage instead of from memory: +// storedMetadataDocument streams it from upstream into storage, and the +// rewrite reads it back as often as it needs. Nothing holds the whole +// document, so its size is not limited by metadata_max_size or by memory. + +// errMetadataNotStored is returned when metadata is not cached in storage, so +// a document has to be read into memory. +var errMetadataNotStored = errors.New("metadata is not cached in storage") + +// errStoredMetadataChanged is returned when a stored document no longer has +// the digest its cache row records: a fetch replaced it while it was read. +var errStoredMetadataChanged = errors.New("stored metadata changed while it was read") + +// storedMetadataDocument returns the cache row of a metadata document stored +// uncompressed with its digest: the stored copy while it is within its TTL, +// or else the document fetched from upstream and stored, or the stored copy +// when upstream fails. The document is streamed from upstream into storage +// without being held in memory, so a fetch takes no metadata_max_size limit. +// Concurrent calls for one document share a fetch, and instances sharing the +// database take turns as for other metadata. +// +// validate, when not nil, checks a fetched document before it replaces the +// stored one; a document it rejects is not stored, and the stored copy is +// served as when upstream fails. It does not check a stored copy within its +// TTL. +// +// It returns errMetadataNotStored when the proxy doesn't cache metadata or +// storing the document failed, and the caller reads it into memory instead. +func (p *Proxy) storedMetadataDocument(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept string, validate func(io.Reader) error) (*database.MetadataCacheEntry, error) { + return p.storedMetadata(ctx, ecosystem, cacheKey, upstreamURL, accept, validate, "") +} + +// restoreStoredMetadata is storedMetadataDocument for a document whose row +// gone describes but whose stored copy was missing when read: the row is not +// served or revalidated, within its TTL or not, and the document is fetched +// again. +func (p *Proxy) restoreStoredMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept string, validate func(io.Reader) error, gone *database.MetadataCacheEntry) (*database.MetadataCacheEntry, error) { + return p.storedMetadata(ctx, ecosystem, cacheKey, upstreamURL, accept, validate, gone.ContentDigest.String) +} + +// storedMetadata implements storedMetadataDocument. A row with the digest +// gone is treated as if there were none. +func (p *Proxy) storedMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept string, validate func(io.Reader) error, gone string) (*database.MetadataCacheEntry, error) { + if !p.CacheMetadata || p.DB == nil || p.Storage == nil || p.MetadataTTL <= 0 { + return nil, errMetadataNotStored + } + if containsPathTraversal(cacheKey) { + return nil, fmt.Errorf("invalid cache key: %q", cacheKey) + } + if entry, fresh := p.storedMetadataEntryState(ctx, ecosystem, cacheKey, gone); fresh { + metrics.RecordMetadataCacheHit(ecosystem, int(entry.Size.Int64)) + p.recordMetadataRead(entry) + return entry, nil + } + p.recordMetadataCacheMiss(ecosystem) + + key := metadataCoalesceKey(ecosystem, cacheKey, upstreamURL, accept, "", false) + "\x00stored" + res, _ := p.coalesceMetadata(ctx, key, func(fetchCtx context.Context) metadataResult { + entry, fresh := p.storedMetadataEntryState(fetchCtx, ecosystem, cacheKey, gone) + if fresh { + return metadataResult{entry: entry} + } + release, outcome, err := p.awaitFetchLease(fetchCtx, metrics.CacheKindMetadata, key, func() bool { + entry, fresh = p.storedMetadataEntryState(fetchCtx, ecosystem, cacheKey, gone) + return fresh + }) + if err != nil { + return metadataResult{err: err} + } + switch outcome { + case leaseStored: + return metadataResult{entry: entry} + case leaseFailed: + // Serve the stored copy, as the holder did, rather than trying + // the failing upstream again. + p.reloadUpstreamHolds() + if usableStoredMetadata(entry) { + return p.staleStoredMetadata(ecosystem, cacheKey, entry, errFetchFailedElsewhere) + } + case leaseFetch: + } + res := metadataResult{err: errSharedFetchAbandoned} + defer func() { release(res.err != nil || res.stale) }() + res = p.fetchStoredMetadata(fetchCtx, ecosystem, cacheKey, upstreamURL, accept, validate, entry) + return res + }) + return res.entry, res.err +} + +// storedMetadataEntryState reads the cache row of a document and reports +// whether it is within its TTL and usable by storedMetadataDocument. A row +// with the digest gone is returned as nil. +func (p *Proxy) storedMetadataEntryState(ctx context.Context, ecosystem, cacheKey, gone string) (*database.MetadataCacheEntry, bool) { + entry, _ := p.DB.GetMetadataCache(ecosystem, cacheKey) + entry = p.migrateLegacyMetadataEntry(ctx, ecosystem, cacheKey, entry) + if gone != "" && entry != nil && entry.ContentDigest.String == gone { + return nil, false + } + fresh := usableStoredMetadata(entry) && entry.FetchedAt.Valid && time.Since(entry.FetchedAt.Time) < p.MetadataTTL + return entry, fresh +} + +// usableStoredMetadata reports whether entry describes a document stored +// uncompressed with its digest, which a rewrite can read back and check. +func usableStoredMetadata(entry *database.MetadataCacheEntry) bool { + return entry != nil && entry.ContentDigest.Valid && entry.ContentEncoding.String == "" +} + +// fetchStoredMetadata fetches a document from upstream into storage, +// revalidating entry when there is one, and falls back to entry when +// upstream fails or validate rejects what it sent. A failure to store is +// returned as it is, for the caller to read the document into memory. +func (p *Proxy) fetchStoredMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept string, validate func(io.Reader) error, entry *database.MetadataCacheEntry) metadataResult { + revalidate := entry + if !usableStoredMetadata(entry) { + revalidate = nil + } + stored, err := p.storeUpstreamMetadata(ctx, ecosystem, cacheKey, upstreamURL, accept, validate, revalidate) + if errors.Is(err, errStale304) { + stored, err = p.storeUpstreamMetadata(ctx, ecosystem, cacheKey, upstreamURL, accept, validate, nil) + } + if err == nil { + return metadataResult{entry: stored} + } + // As in fetchMetadataFromUpstream, any upstream failure, a 404 included, + // falls back to the stored copy. + if !usableStoredMetadata(entry) || errors.Is(err, errMetadataNotStored) { + return metadataResult{err: err} + } + return p.staleStoredMetadata(ecosystem, cacheKey, entry, err) +} + +// staleStoredMetadata falls back to the stored document entry describes after +// the upstream fetch failed with err. +func (p *Proxy) staleStoredMetadata(ecosystem, cacheKey string, entry *database.MetadataCacheEntry, err error) metadataResult { + p.Logger.Warn("upstream metadata fetch failed, serving cached copy", + "ecosystem", ecosystem, "key", cacheKey, "error", err) + entry = p.currentMetadataEntry(ecosystem, cacheKey, entry) + if !usableStoredMetadata(entry) { + return metadataResult{err: fmt.Errorf("upstream failed and no usable cached metadata: %w", err)} + } + return metadataResult{entry: entry, stale: true} +} + +// storeUpstreamMetadata fetches a document and streams it into storage, and +// returns its new cache row. On a 304 for entry it only moves the row's +// fetched_at on; errStale304 means the stored copy is gone or the row was +// replaced meanwhile, and the caller should fetch without validators. +// +// With validate, the document is staged under a path of its own and checked +// there first, so one validate rejects never replaces the stored copy. +func (p *Proxy) storeUpstreamMetadata(ctx context.Context, ecosystem, cacheKey, upstreamURL, accept string, validate func(io.Reader) error, entry *database.MetadataCacheEntry) (*database.MetadataCacheEntry, error) { + resp, err := p.requestUpstreamMetadata(ctx, upstreamURL, entry, accept, "") + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode == http.StatusNotModified && entry != nil { + if ok, err := p.Storage.Exists(ctx, entry.StoragePath); err != nil || !ok { + return nil, errStale304 + } + refreshed, err := p.DB.RefreshMetadataCache(ecosystem, cacheKey, entry.ContentDigest.String, time.Now()) + if err != nil || !refreshed { + return nil, errStale304 + } + return p.currentMetadataEntry(ecosystem, cacheKey, entry), nil + } + if err := upstreamMetadataStatus(resp); err != nil { + return nil, err + } + + meta := &upstreamMetadata{ + contentType: resp.Header.Get(headerContentType), + etag: resp.Header.Get(headerETag), + } + if meta.contentType == "" { + meta.contentType = contentTypeJSON + } + if lm := resp.Header.Get(headerLastModified); lm != "" { + meta.lastModified, _ = http.ParseTime(lm) + } + limit := p.MetadataStreamMaxSize + if limit <= 0 { + limit = defaultMetadataStreamMaxSize + } + // Go decompresses the body, since no Accept-Encoding was set, so it is + // stored uncompressed. + var body io.Reader = &limitedBody{r: upstreamBodyReader{resp.Body}, left: limit} + if validate != nil { + staged, err := p.stageMetadata(ctx, ecosystem, cacheKey, body, validate) + if err != nil { + return nil, err + } + defer func() { _ = staged.Close() }() + body = staged + } + if _, err := p.storeMetadataBlob(ctx, ecosystem, cacheKey, metadataStoragePath(ecosystem, cacheKey), body, meta); err != nil { + return nil, notStored("", err) + } + stored, err := p.DB.GetMetadataCache(ecosystem, cacheKey) + if err != nil || !usableStoredMetadata(stored) { + return nil, fmt.Errorf("reading the stored metadata row: %w", errors.Join(err, errMetadataNotStored)) + } + return stored, nil +} + +// stageMetadata stores body under a staging path of its own, checks it with +// validate, and returns it opened for reading, to be stored in place of the +// cached copy. The staged copy is deleted when the reader is closed. A +// failure to store it is returned as errMetadataNotStored. +func (p *Proxy) stageMetadata(ctx context.Context, ecosystem, cacheKey string, body io.Reader, validate func(io.Reader) error) (io.ReadCloser, error) { + path := metadataStoragePath(ecosystem, cacheKey) + ".staged-" + rand.Text() + discard := func() { + if err := p.Storage.Delete(context.WithoutCancel(ctx), path); err != nil { + p.Logger.Warn("failed to delete staged metadata", "ecosystem", ecosystem, "key", cacheKey, "error", err) + } + } + if _, _, err := p.Storage.Store(ctx, path, body); err != nil { + discard() + return nil, notStored("staging: ", err) + } + if err := readStored(func() (io.ReadCloser, error) { return p.Storage.Open(ctx, path) }, validate); err != nil { + discard() + return nil, err + } + staged, err := p.Storage.Open(ctx, path) + if err != nil { + discard() + return nil, fmt.Errorf("%w: reading staged copy: %w", errMetadataNotStored, err) + } + return &stagedReader{ReadCloser: staged, discard: discard}, nil +} + +// limitedBody fails with ErrMetadataTooLarge, as an upstream failure, once +// r has sent more than left bytes. +type limitedBody struct { + r io.Reader + left int64 +} + +func (l *limitedBody) Read(b []byte) (int, error) { + n, err := l.r.Read(b) + if l.left -= int64(n); l.left < 0 { + return 0, &upstreamBodyError{ErrMetadataTooLarge} + } + return n, err +} + +// notStored returns a failure to store a document as errMetadataNotStored, +// unless reading it from upstream failed. +func notStored(what string, err error) error { + if body := (*upstreamBodyError)(nil); errors.As(err, &body) { + return body + } + return fmt.Errorf("%w: %s%w", errMetadataNotStored, what, err) +} + +// stagedReader deletes the staged copy it reads when closed. +type stagedReader struct { + io.ReadCloser + discard func() +} + +func (s *stagedReader) Close() error { + err := s.ReadCloser.Close() + s.discard() + return err +} + +// openStoredMetadata opens the document entry describes. Reading it to the +// end fails with errStoredMetadataChanged when its bytes don't have the +// digest entry records. +func (p *Proxy) openStoredMetadata(ctx context.Context, entry *database.MetadataCacheEntry) (io.ReadCloser, error) { + r, err := p.Storage.Open(ctx, entry.StoragePath) + if err != nil { + return nil, err + } + return &digestReader{r: r, h: sha256.New(), want: entry.ContentDigest.String}, nil +} + +// digestReader checks at the end of its reader that what it read has the +// digest want, in the form the metadata cache records it. +type digestReader struct { + r io.ReadCloser + h hash.Hash + want string +} + +func (d *digestReader) Read(b []byte) (int, error) { + n, err := d.r.Read(b) + d.h.Write(b[:n]) + if errors.Is(err, io.EOF) && "sha256:"+hex.EncodeToString(d.h.Sum(nil)) != d.want { + return n, errStoredMetadataChanged + } + return n, err +} + +func (d *digestReader) Close() error { return d.r.Close() } diff --git a/internal/handler/npm.go b/internal/handler/npm.go index 62bf4409..50757a37 100644 --- a/internal/handler/npm.go +++ b/internal/handler/npm.go @@ -2,6 +2,7 @@ package handler import ( "bytes" + "context" "errors" "fmt" "io" @@ -12,9 +13,11 @@ import ( "time" "github.com/git-pkgs/integrity" + "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/jsonscan" "github.com/git-pkgs/proxy/internal/metadata/npm" "github.com/git-pkgs/proxy/internal/retention" + "github.com/git-pkgs/proxy/internal/storage" ) const ( @@ -264,6 +267,10 @@ func (h *NPMHandler) handlePackageMetadata(w http.ResponseWriter, r *http.Reques return } + if !versionLookup && h.serveStreamedPackument(w, r, packageName, upstreamURL, accept, metadataCacheKey) { + return + } + var validate func([]byte) error if !versionLookup { validate = h.packumentValidator() @@ -317,6 +324,207 @@ func (h *NPMHandler) handlePackageMetadata(w http.ResponseWriter, r *http.Reques serveRewrittenMetadata(w, r, rewritten) } +// serveStreamedPackument serves a packument rewritten from storage: the +// document is streamed from upstream into storage, then read twice, once to +// collect the time map and version names cooldown and the denylist decide +// on, and once to copy each kept version. registry.npmjs.org writes +// dist-tags before versions and time after them, so one pass can't do it. +// Neither the packument nor its rewrite is held in memory uncompressed, and +// packuments larger than metadata_max_size can be served. +// +// It reports false when it leaves the request to the in-memory rewrite: +// without metadata or rewrite caching, for a document it can't stream, such +// as an abbreviated packument where the full one is required, and when the +// rewrite fails. The document is stored by then, so the in-memory path reads +// it from storage. +func (h *NPMHandler) serveStreamedPackument(w http.ResponseWriter, r *http.Request, packageName, upstreamURL, accept, metadataCacheKey string) bool { + if h.proxy.rewrites == nil { + return false + } + fullRequired := h.packumentValidator() != nil + options := h.npmRewriteOptions(packageName, nil) + var rewritten rewrittenMetadata + err, fetchErr := h.readStoredPackument(r.Context(), upstreamURL, accept, metadataCacheKey, func(entry *database.MetadataCacheEntry) (err error) { + rewritten, err = h.proxy.streamedRewrite(r.Context(), "npm", h.proxyURL, metadataCacheKey, entry, func(open func() (io.ReadCloser, error), out io.Writer) error { + var scan *npm.Scan + err := readStored(open, func(in io.Reader) (err error) { + scan, err = npm.ScanPackument(in) + return err + }) + if err != nil { + return err + } + if fullRequired && scan.Abbreviated() { + return npm.ErrNotStreamable + } + return readStored(open, func(in io.Reader) error { + _, err := npm.RewriteStream(packageName, scan, in, out, options) + return err + }) + }) + return err + }) + switch { + case errors.Is(fetchErr, errInMemory): + return false + case fetchErr != nil: + h.writeFetchError(w, fetchErr) + return true + } + if err != nil { + if r.Context().Err() != nil { + return true // the client left while waiting on a shared rewrite + } + if !errors.Is(err, npm.ErrNotStreamable) { + h.proxy.Logger.Warn("failed to rewrite stored npm metadata, rewriting in memory", + "package", packageName, "error", err) + } + return false + } + serveRewrittenMetadata(w, r, rewritten) + return true +} + +// npmFetchError is a failure to fetch a packument, answered as such rather +// than as a failure to filter it. +type npmFetchError struct{ err error } + +func (e *npmFetchError) Error() string { return e.err.Error() } +func (e *npmFetchError) Unwrap() error { return e.err } + +// writeFetchError answers a request whose packument could not be fetched. +func (h *NPMHandler) writeFetchError(w http.ResponseWriter, err error) { + switch { + case errors.Is(err, ErrUpstreamNotFound): + JSONError(w, http.StatusNotFound, ErrPackageNotFound.Error()) + case writeUpstreamHeld(w, err): + default: + h.proxy.Logger.Error("failed to fetch npm metadata", "error", err) + JSONError(w, http.StatusBadGateway, "failed to fetch from upstream") + } +} + +// errInMemory is returned when a packument has to be read into memory to +// answer the request: metadata is not cached in storage, or the stored +// document can't be read as a stream. +var errInMemory = errors.New("npm packument needs the in-memory path") + +// storedPackument returns the cache row of the packument stored for cacheKey, +// fetching it into storage first when needed. It returns errInMemory when +// metadata isn't cached in storage, and ErrUpstreamNotFound and upstream +// errors as they come. +func (h *NPMHandler) storedPackument(ctx context.Context, packumentURL, accept, cacheKey string) (*database.MetadataCacheEntry, error) { + entry, err := h.proxy.storedMetadataDocument(ctx, "npm", cacheKey, packumentURL, accept, h.packumentStreamValidator()) + if errors.Is(err, errMetadataNotStored) { + return nil, errInMemory + } + return entry, err +} + +// readStoredPackument reads the packument stored for cacheKey with read, +// fetching it into storage first when needed. When the stored copy is gone +// or no longer has the digest its row records, the packument is fetched +// again and read once more. fetchErr is what storedPackument returned when the packument +// couldn't be fetched, and err is what read returned. +func (h *NPMHandler) readStoredPackument(ctx context.Context, packumentURL, accept, cacheKey string, read func(*database.MetadataCacheEntry) error) (err, fetchErr error) { + entry, fetchErr := h.storedPackument(ctx, packumentURL, accept, cacheKey) + if fetchErr != nil { + return nil, fetchErr + } + if err = read(entry); !errors.Is(err, storage.ErrNotFound) && !errors.Is(err, errStoredMetadataChanged) { + return err, nil + } + h.proxy.Logger.Warn("stored npm metadata is missing or changed, fetching it again", "url", packumentURL, "error", err) + entry, fetchErr = h.proxy.restoreStoredMetadata(ctx, "npm", cacheKey, packumentURL, accept, h.packumentStreamValidator(), entry) + if errors.Is(fetchErr, errMetadataNotStored) { + return nil, errInMemory + } + if fetchErr != nil { + return nil, fetchErr + } + return read(entry), nil +} + +// packumentStreamValidator is packumentValidator for a packument read from +// storage. +func (h *NPMHandler) packumentStreamValidator() func(io.Reader) error { + if h.packumentValidator() == nil { + return nil + } + return func(r io.Reader) error { + entry, err := npm.LookupVersion(r, "") + if err != nil { + return err + } + if entry.Abbreviated { + return errAbbreviatedPackument + } + return nil + } +} + +// streamedVersionMetadata resolves a version or dist-tag through the +// packument stored in storage, as rewriteNPMVersionFromPackument does in +// memory: one pass collects what cooldown and the denylist decide on, and a +// second finds the version entry. It returns errInMemory when the request +// has to take the in-memory path. +func (h *NPMHandler) streamedVersionMetadata(ctx context.Context, packageName, specifier, packumentURL, accept, cacheKey, metadataCacheKey string) (rewrittenMetadata, error) { + if h.proxy.rewrites == nil { + return rewrittenMetadata{}, errInMemory + } + fullRequired := h.packumentValidator() != nil + options := h.npmRewriteOptions(packageName, nil) + var m rewrittenMetadata + err, fetchErr := h.readStoredPackument(ctx, packumentURL, accept, cacheKey, func(entry *database.MetadataCacheEntry) (err error) { + m, err = h.proxy.streamedRewrite(ctx, "npm", h.proxyURL, metadataCacheKey, entry, func(open func() (io.ReadCloser, error), out io.Writer) error { + var scan *npm.Scan + err := readStored(open, func(in io.Reader) (err error) { + scan, err = npm.ScanPackument(in) + return err + }) + if err != nil { + return err + } + if fullRequired && scan.Abbreviated() { + return npm.ErrNotStreamable + } + return readStored(open, func(in io.Reader) error { + version, err := npm.VersionStream(packageName, specifier, scan, in, options) + if err != nil { + return npmRewriteError(err) + } + _, err = out.Write(version) + return err + }) + }) + return err + }) + switch { + case errors.Is(fetchErr, errInMemory): + return rewrittenMetadata{}, fetchErr + case fetchErr != nil: + return rewrittenMetadata{}, &npmFetchError{fetchErr} + } + switch { + case err == nil, ctx.Err() != nil, errors.Is(err, ErrVersionDenied), errors.Is(err, ErrVersionNotFound): + return m, err + case !errors.Is(err, npm.ErrNotStreamable): + h.proxy.Logger.Warn("failed to read stored npm metadata, reading it in memory", + "package", packageName, "error", err) + } + return rewrittenMetadata{}, errInMemory +} + +// readStored opens a stored document and reads it with read. +func readStored(open func() (io.ReadCloser, error), read func(io.Reader) error) error { + in, err := open() + if err != nil { + return err + } + defer func() { _ = in.Close() }() + return read(in) +} + // packumentAccept returns the Accept header packuments are requested with. // It prefers the smaller abbreviated packument format but includes // application/json as a fallback so upstreams that reject the abbreviated type @@ -371,7 +579,19 @@ func requireFullPackument(body []byte) error { // filtered packument so cooldown and dist-tags match the root metadata response. func (h *NPMHandler) handleFilteredNPMVersionMetadata(w http.ResponseWriter, r *http.Request, packageName, specifier, accept, registryURL, cacheKey, metadataCacheKey string) { _, packumentPath := npmMetadataPath(packageName) - packument, _, _, err := h.proxy.fetchOrCacheMetadata(r.Context(), "npm", cacheKey, registryURL+"/"+packumentPath, "", h.packumentValidator(), accept) + packumentURL := registryURL + "/" + packumentPath + rewritten, err := h.streamedVersionMetadata(r.Context(), packageName, specifier, packumentURL, accept, cacheKey, metadataCacheKey) + var fetchErr *npmFetchError + switch { + case errors.As(err, &fetchErr): + h.writeFetchError(w, fetchErr.err) + return + case !errors.Is(err, errInMemory): + h.writeVersionMetadata(w, r, packageName, rewritten, err) + return + } + + packument, _, _, err := h.proxy.fetchOrCacheMetadata(r.Context(), "npm", cacheKey, packumentURL, "", h.packumentValidator(), accept) if err != nil { if errors.Is(err, ErrUpstreamNotFound) { JSONError(w, http.StatusNotFound, ErrPackageNotFound.Error()) @@ -385,10 +605,16 @@ func (h *NPMHandler) handleFilteredNPMVersionMetadata(w http.ResponseWriter, r * return } - rewritten, err := h.proxy.cachedRewrite(r.Context(), "npm", h.proxyURL, metadataCacheKey, packument, "", func(b []byte) ([]byte, []string, error) { + rewritten, err = h.proxy.cachedRewrite(r.Context(), "npm", h.proxyURL, metadataCacheKey, packument, "", func(b []byte) ([]byte, []string, error) { out, err := h.rewriteNPMVersionFromPackument(packageName, specifier, b) return out, nil, err }) + h.writeVersionMetadata(w, r, packageName, rewritten, err) +} + +// writeVersionMetadata answers a version metadata request with its rewrite +// or the error that kept it from being made. +func (h *NPMHandler) writeVersionMetadata(w http.ResponseWriter, r *http.Request, packageName string, rewritten rewrittenMetadata, err error) { if err != nil { if r.Context().Err() != nil { return @@ -549,10 +775,8 @@ func (h *NPMHandler) handleDownload(w http.ResponseWriter, r *http.Request) { return } registryURL, cacheKey := h.sourceFor(packageName) - metadata := sync.OnceValues(func() ([]byte, error) { - upstreamURL := registryURL + "/" + url.PathEscape(packageName) - body, _, _, err := h.proxy.fetchOrCacheMetadata(r.Context(), "npm", cacheKey, upstreamURL, "", h.packumentValidator(), h.packumentAccept()) - return body, err + metadata := sync.OnceValues(func() (npmVersionData, error) { + return h.versionData(r.Context(), registryURL+"/"+url.PathEscape(packageName), cacheKey, version) }) inCooldown, err := h.versionInCooldown(packageName, version, metadata) if err != nil { @@ -594,7 +818,7 @@ func (h *NPMHandler) handleDownload(w http.ResponseWriter, r *http.Request) { ServeArtifactRequest(w, r, result) } -func (h *NPMHandler) getTarball(r *http.Request, registryURL, packageName, version, filename string, metadata func() ([]byte, error)) (*CacheResult, error) { +func (h *NPMHandler) getTarball(r *http.Request, registryURL, packageName, version, filename string, metadata func() (npmVersionData, error)) (*CacheResult, error) { cacheFilename := h.routes.cacheFilename(packageName, filename) if cached, err := h.proxy.GetCachedArtifact(r.Context(), "npm", packageName, version, cacheFilename); err != nil || cached != nil { return cached, err @@ -602,35 +826,35 @@ func (h *NPMHandler) getTarball(r *http.Request, registryURL, packageName, versi // The packument says where the tarball is and what it must hash to. // Without it the tarball could not be verified, and an unverified copy // would be cached and served to everyone, so the download fails instead. - body, err := metadata() + data, err := metadata() if err != nil { return nil, fmt.Errorf("fetching npm metadata to verify tarball: %w", err) } downloadURL := fmt.Sprintf("%s/%s/-/%s", registryURL, escapeNPMDownloadPackage(packageName), url.PathEscape(filename)) - if tarball := npmVersionTarball(body, version); tarball != "" { + if tarball := npmEntryTarball(data.entry); tarball != "" { downloadURL, err = validateTarballURL(tarball, registryURL) if err != nil { return nil, err } } - sri := npmVersionIntegrity(body, version) + sri := npmEntryIntegrity(data.entry) return h.proxy.GetOrFetchArtifactFromURLWithIntegrity(r.Context(), "npm", packageName, version, cacheFilename, downloadURL, sri) } -// npmVersionIntegrity returns the integrity a packument declares for a -// version's tarball: dist.integrity when it is a Subresource Integrity string +// npmEntryIntegrity returns the integrity a version's packument entry +// declares for its tarball: dist.integrity when it is a Subresource Integrity string // the proxy can verify, or else the SHA-1 in dist.shasum, which is all // versions published before integrity existed carry, as "sha1-". It // returns "" when neither is usable, and the tarball is fetched unverified as // before. -func npmVersionIntegrity(body []byte, version string) string { - sri, _, _ := jsonscan.LookupString(body, "versions", version, "dist", "integrity") +func npmEntryIntegrity(entry []byte) string { + sri, _, _ := jsonscan.LookupString(entry, "dist", "integrity") if sri != "" { if _, err := integrity.ParseSRI(sri); err == nil { return sri } } - shasum, _, _ := jsonscan.LookupString(body, "versions", version, "dist", "shasum") + shasum, _, _ := jsonscan.LookupString(entry, "dist", "shasum") return sha1Checksum(shasum) } @@ -638,10 +862,57 @@ func npmVersionIntegrity(body []byte, version string) string { // or "" if it has none. It reads that one value in place rather than decoding // every version, since it runs on each tarball the proxy has not cached yet. func npmVersionTarball(body []byte, version string) string { - tarball, _, _ := jsonscan.LookupString(body, "versions", version, "dist", "tarball") + entry, _ := jsonscan.Lookup(body, "versions", version) + return npmEntryTarball(entry) +} + +// npmEntryTarball is npmVersionTarball for the version's entry. +func npmEntryTarball(entry []byte) string { + tarball, _, _ := jsonscan.LookupString(entry, "dist", "tarball") return tarball } +// npmVersionData is what a tarball download reads from the packument: the +// version's entry, nil when it has none, and its publish time from the time +// map, "" when it has none. +type npmVersionData struct { + entry []byte + published string +} + +// versionData reads what a tarball download needs about version from the +// packument: from storage in one pass when metadata is cached there, and +// otherwise from the packument read into memory. +func (h *NPMHandler) versionData(ctx context.Context, packumentURL, cacheKey, version string) (npmVersionData, error) { + accept := h.packumentAccept() + var found npm.VersionEntry + err, fetchErr := h.readStoredPackument(ctx, packumentURL, accept, cacheKey, func(entry *database.MetadataCacheEntry) error { + return readStored(func() (io.ReadCloser, error) { return h.proxy.openStoredMetadata(ctx, entry) }, func(in io.Reader) (err error) { + found, err = npm.LookupVersion(in, version) + return err + }) + }) + switch { + case fetchErr == nil && err == nil && (!found.Abbreviated || h.packumentValidator() == nil): + return npmVersionData{entry: found.Entry, published: found.Published}, nil + case fetchErr == nil && err != nil: + h.proxy.Logger.Warn("failed to read stored npm metadata, reading it in memory", "url", packumentURL, "error", err) + case fetchErr != nil && !errors.Is(fetchErr, errInMemory): + return npmVersionData{}, fetchErr + } + + body, _, _, err := h.proxy.fetchOrCacheMetadata(ctx, "npm", cacheKey, packumentURL, "", h.packumentValidator(), accept) + if err != nil { + return npmVersionData{}, err + } + entry, _ := jsonscan.Lookup(body, "versions", version) + published, _, err := jsonscan.LookupString(body, "time", version) + if err != nil { + return npmVersionData{}, fmt.Errorf("parsing npm metadata: %w", err) + } + return npmVersionData{entry: entry, published: published}, nil +} + func validateTarballURL(raw, registryURL string) (string, error) { return validateRegistryURL("tarball", raw, registryURL) } @@ -692,7 +963,7 @@ func validateRegistryURL(what, raw, registryURL string) (string, error) { // for a private version of the same name, a private time would hold back the // public version, and moving a route to another upstream would keep the old // one's times. -func (h *NPMHandler) versionInCooldown(packageName, version string, metadata func() ([]byte, error)) (bool, error) { +func (h *NPMHandler) versionInCooldown(packageName, version string, metadata func() (npmVersionData, error)) (bool, error) { if !h.proxy.cooldownApplies("npm", packageName) { return false, nil } @@ -705,7 +976,7 @@ func (h *NPMHandler) versionInCooldown(packageName, version string, metadata fun } } - body, err := metadata() + data, err := metadata() if errors.Is(err, ErrUpstreamNotFound) { return false, nil } @@ -713,11 +984,8 @@ func (h *NPMHandler) versionInCooldown(packageName, version string, metadata fun return false, err } - published, ok, err := jsonscan.LookupString(body, "time", version) - if err != nil { - return false, fmt.Errorf("parsing npm metadata: %w", err) - } - if !ok { + published := data.published + if published == "" { return false, nil } diff --git a/internal/handler/npm_attestations.go b/internal/handler/npm_attestations.go index 2694af24..02c57513 100644 --- a/internal/handler/npm_attestations.go +++ b/internal/handler/npm_attestations.go @@ -40,10 +40,8 @@ func (h *NPMHandler) handleAttestations(w http.ResponseWriter, r *http.Request) } registryURL, cacheKey := h.sourceFor(packageName) - metadata := sync.OnceValues(func() ([]byte, error) { - upstreamURL := registryURL + "/" + url.PathEscape(packageName) - body, _, _, err := h.proxy.fetchOrCacheMetadata(r.Context(), "npm", cacheKey, upstreamURL, "", h.packumentValidator(), h.packumentAccept()) - return body, err + metadata := sync.OnceValues(func() (npmVersionData, error) { + return h.versionData(r.Context(), registryURL+"/"+url.PathEscape(packageName), cacheKey, version) }) inCooldown, err := h.versionInCooldown(packageName, version, metadata) @@ -61,7 +59,7 @@ func (h *NPMHandler) handleAttestations(w http.ResponseWriter, r *http.Request) return } - packument, err := metadata() + data, err := metadata() if err != nil { if errors.Is(err, ErrUpstreamNotFound) { JSONError(w, http.StatusNotFound, ErrPackageNotFound.Error()) @@ -74,7 +72,7 @@ func (h *NPMHandler) handleAttestations(w http.ResponseWriter, r *http.Request) JSONError(w, http.StatusBadGateway, "failed to fetch from upstream") return } - listed, _, _ := jsonscan.LookupString(packument, "versions", version, "dist", "attestations", "url") + listed, _, _ := jsonscan.LookupString(data.entry, "dist", "attestations", "url") if listed == "" { JSONError(w, http.StatusNotFound, errNoAttestations.Error()) return diff --git a/internal/handler/npm_stream_harden_test.go b/internal/handler/npm_stream_harden_test.go new file mode 100644 index 00000000..0cebf0a7 --- /dev/null +++ b/internal/handler/npm_stream_harden_test.go @@ -0,0 +1,238 @@ +package handler + +import ( + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +// bigMetadataFiles returns a copy of what is in storage under the metadata path +// of big, staged copies included. +func bigMetadataFiles(store *mockStorage) map[string]string { + store.mu.Lock() + defer store.mu.Unlock() + files := map[string]string{} + for path, b := range store.files { + if strings.HasPrefix(path, metadataStoragePath("npm", "big")) { + files[path] = string(b) + } + } + return files +} + +// TestNPMStreamedPackumentBlobCorrupt checks that a stored copy whose bytes +// no longer match the digest of its row, within its TTL, is fetched again +// rather than failing the request, also above metadata_max_size. +func TestNPMStreamedPackumentBlobCorrupt(t *testing.T) { + body := bigPackument(400) + u := newStreamUpstream(t, body) + proxy, store, handler := streamProxy(t, u) + proxy.MetadataMaxSize = int64(len(body) / 4) + decodeStreamed(t, serve(handler, "/big", nil)) + + path := metadataStoragePath("npm", "big") + store.mu.Lock() + store.files[path] = []byte(strings.Replace(body, "1.0.5", "1.0.X", 1)) + store.mu.Unlock() + proxy.SetMetadataRewriteCacheSize(64 << 20) + + if doc := decodeStreamed(t, serve(handler, "/big", nil)); len(doc.Versions) != 400 { + t.Errorf("served %d versions, want 400", len(doc.Versions)) + } + if got := bigMetadataFiles(store)[path]; got != body { + t.Error("the corrupt copy was not replaced") + } +} + +// TestNPMStreamedPackumentTruncated checks that a document upstream cuts +// off halfway neither replaces the stored copy, which is served instead, +// nor leaves anything behind, with and without cooldown. +func TestNPMStreamedPackumentTruncated(t *testing.T) { + for _, withCooldown := range []bool{true, false} { + t.Run(map[bool]string{true: "cooldown", false: "no cooldown"}[withCooldown], func(t *testing.T) { + body := bigPackument(400) + u := newStreamUpstream(t, body) + proxy, store, handler := streamProxy(t, u) + if !withCooldown { + proxy.Cooldown = nil + } + proxy.MetadataMaxSize = int64(len(body) / 4) + want := len(decodeStreamed(t, serve(handler, "/big", nil)).Versions) + before := bigMetadataFiles(store) + + u.replace(bigPackument(401), `"v2"`) + u.truncate.Store(true) + proxy.MetadataTTL = time.Nanosecond + proxy.SetMetadataRewriteCacheSize(64 << 20) + + if doc := decodeStreamed(t, serve(handler, "/big", nil)); len(doc.Versions) != want { + t.Errorf("served %d versions, want the stored %d", len(doc.Versions), want) + } + after := bigMetadataFiles(store) + if len(after) != len(before) { + t.Errorf("storage holds %d files, %d before", len(after), len(before)) + } + for path, b := range before { + if after[path] != b { + t.Errorf("%s was changed", path) + } + } + }) + } +} + +// TestNPMStreamedPackumentTruncatedUncached checks that a document upstream +// cuts off halfway fails the request and is not kept. +func TestNPMStreamedPackumentTruncatedUncached(t *testing.T) { + body := bigPackument(400) + u := newStreamUpstream(t, body) + proxy, store, handler := streamProxy(t, u) + proxy.Cooldown = nil + proxy.MetadataMaxSize = int64(len(body) / 4) + u.truncate.Store(true) + + if rec := serve(handler, "/big", nil); rec.Code != http.StatusBadGateway { + t.Errorf("status %d, want 502", rec.Code) + } + if files := bigMetadataFiles(store); len(files) != 0 { + t.Errorf("left %d files in storage", len(files)) + } + if entry, _ := proxy.DB.GetMetadataCache("npm", "big"); entry != nil { + t.Error("left a metadata cache row") + } +} + +// TestNPMStreamedPackumentUncachedErrors checks the answers to upstream +// failures with nothing cached. +func TestNPMStreamedPackumentUncachedErrors(t *testing.T) { + for _, tt := range []struct { + name string + set func(*streamUpstream) + want int + }{ + {"not found", func(u *streamUpstream) { u.gone.Store(true) }, http.StatusNotFound}, + {"upstream down", func(u *streamUpstream) { u.fail.Store(true) }, http.StatusBadGateway}, + } { + t.Run(tt.name, func(t *testing.T) { + u := newStreamUpstream(t, bigPackument(10)) + _, store, handler := streamProxy(t, u) + tt.set(u) + if rec := serve(handler, "/big", nil); rec.Code != tt.want { + t.Errorf("status %d, want %d", rec.Code, tt.want) + } + if files := bigMetadataFiles(store); len(files) != 0 { + t.Errorf("left %d files in storage", len(files)) + } + }) + } +} + +// TestNPMStreamedPackumentConcurrent checks that concurrent requests for an +// uncached packument share one upstream fetch and all get it. +func TestNPMStreamedPackumentConcurrent(t *testing.T) { + body := bigPackument(400) + u := newStreamUpstream(t, body) + proxy, _, handler := streamProxy(t, u) + proxy.MetadataMaxSize = int64(len(body) / 4) + const n = 16 + var wg sync.WaitGroup + recs := make([]*httptest.ResponseRecorder, n) + for i := range n { + wg.Go(func() { recs[i] = serve(handler, "/big", nil) }) + } + wg.Wait() + for _, rec := range recs { + if doc := decodeStreamed(t, rec); len(doc.Versions) != 400 { + t.Errorf("served %d versions, want 400", len(doc.Versions)) + } + } + if got := u.sent.Load(); got != 1 { + t.Errorf("upstream sent the document %d times, want once", got) + } +} + +// TestNPMStreamedPackumentClientRevalidates checks that a client sending +// the ETag it got gets a 304, and a new ETag once the document changes. +func TestNPMStreamedPackumentClientRevalidates(t *testing.T) { + body := bigPackument(400) + u := newStreamUpstream(t, body) + proxy, _, handler := streamProxy(t, u) + proxy.MetadataMaxSize = int64(len(body) / 4) + + first := serve(handler, "/big", nil) + etag := first.Header().Get("ETag") + if first.Code != http.StatusOK || etag == "" { + t.Fatalf("status %d, ETag %q", first.Code, etag) + } + if rec := serve(handler, "/big", http.Header{"If-None-Match": {etag}}); rec.Code != http.StatusNotModified { + t.Errorf("same document: status %d, want 304", rec.Code) + } + + u.replace(bigPackument(401), `"v2"`) + proxy.MetadataTTL = time.Nanosecond + rec := serve(handler, "/big", http.Header{"If-None-Match": {etag}}) + if rec.Code != http.StatusOK || rec.Header().Get("ETag") == etag { + t.Errorf("changed document: status %d, ETag %q (was %q)", rec.Code, rec.Header().Get("ETag"), etag) + } + if doc := decodeStreamed(t, rec); len(doc.Versions) != 401 { + t.Errorf("served %d versions, want 401", len(doc.Versions)) + } +} + +// TestNPMStreamedPackumentOverStreamLimit checks that a document over +// MetadataStreamMaxSize is neither stored nor served: the stored copy is +// served instead, with and without cooldown, and with nothing stored the +// request fails. +func TestNPMStreamedPackumentOverStreamLimit(t *testing.T) { + for _, withCooldown := range []bool{true, false} { + t.Run(map[bool]string{true: "cooldown", false: "no cooldown"}[withCooldown], func(t *testing.T) { + body := bigPackument(400) + u := newStreamUpstream(t, body) + proxy, store, handler := streamProxy(t, u) + if !withCooldown { + proxy.Cooldown = nil + } + proxy.MetadataMaxSize = int64(len(body) / 4) + proxy.MetadataStreamMaxSize = int64(len(body) + 1000) + want := len(decodeStreamed(t, serve(handler, "/big", nil)).Versions) + before := bigMetadataFiles(store) + + u.replace(bigPackument(500), `"v2"`) + proxy.MetadataTTL = time.Nanosecond + proxy.SetMetadataRewriteCacheSize(64 << 20) + + if doc := decodeStreamed(t, serve(handler, "/big", nil)); len(doc.Versions) != want { + t.Errorf("served %d versions, want the stored %d", len(doc.Versions), want) + } + after := bigMetadataFiles(store) + if len(after) != len(before) { + t.Errorf("storage holds %d files, %d before", len(after), len(before)) + } + for path, b := range before { + if after[path] != b { + t.Errorf("%s was changed", path) + } + } + }) + } + t.Run("uncached", func(t *testing.T) { + body := bigPackument(400) + u := newStreamUpstream(t, body) + proxy, store, handler := streamProxy(t, u) + proxy.MetadataMaxSize = int64(len(body) / 4) + proxy.MetadataStreamMaxSize = int64(len(body) / 2) + + if rec := serve(handler, "/big", nil); rec.Code != http.StatusBadGateway { + t.Errorf("status %d, want 502", rec.Code) + } + if files := bigMetadataFiles(store); len(files) != 0 { + t.Errorf("left %d files in storage", len(files)) + } + if entry, _ := proxy.DB.GetMetadataCache("npm", "big"); entry != nil { + t.Error("left a metadata cache row") + } + }) +} diff --git a/internal/handler/npm_stream_test.go b/internal/handler/npm_stream_test.go new file mode 100644 index 00000000..a8a4ea2a --- /dev/null +++ b/internal/handler/npm_stream_test.go @@ -0,0 +1,470 @@ +package handler + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/cooldown" + "github.com/git-pkgs/registries/fetch" +) + +// bigPackument returns a packument laid out as registry.npmjs.org writes +// them, dist-tags before versions and time after, with n old versions padded +// so the document is large, and a latest published an hour ago. +func bigPackument(n int) string { + return bigPackumentAt("https://registry.npmjs.org", n) +} + +// bigPackumentAt is bigPackument with tarballs under base. +func bigPackumentAt(base string, n int) string { + var b strings.Builder + b.WriteString(`{"_id":"big","name":"big","dist-tags":{"latest":"9.0.0","old":"1.0.0"},"versions":{`) + times := []string{`"created":"2015-01-01T00:00:00Z"`} + for i := range n { + v := fmt.Sprintf("1.0.%d", i) + if i == 0 { + v = "1.0.0" + } + fmt.Fprintf(&b, `%q:{"name":"big","version":%q,"readme":%q,"dist":{"tarball":"%s/big/-/big-%s.tgz"}},`, + v, v, strings.Repeat("x", 512), base, v) + times = append(times, fmt.Sprintf(`%q:"2016-01-01T00:%02d:%02dZ"`, v, i/60%60, i%60)) + } + recent := time.Now().Add(-time.Hour).UTC().Format(time.RFC3339) + b.WriteString(`"9.0.0":{"name":"big","version":"9.0.0","dist":{"tarball":"` + base + `/big/-/big-9.0.0.tgz"}}},`) + times = append(times, `"9.0.0":"`+recent+`"`) + b.WriteString(`"time":{` + strings.Join(times, ",") + `},"readme":"big"}`) + return b.String() +} + +type streamUpstream struct { + *httptest.Server + body atomic.Value // string + etag atomic.Value // string + fail atomic.Bool + gone atomic.Bool + // truncate makes upstream close the connection halfway through the + // document it announced. + truncate atomic.Bool + requests atomic.Int32 + // sent counts the responses that carried the document. + sent atomic.Int32 +} + +// replace makes upstream send body under a new ETag, so a revalidation +// gets it in a 200. +func (u *streamUpstream) replace(body, etag string) { + u.body.Store(body) + u.etag.Store(etag) +} + +func newStreamUpstream(t *testing.T, body string) *streamUpstream { + t.Helper() + u := &streamUpstream{} + u.replace(body, `"v1"`) + u.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + u.requests.Add(1) + switch { + case u.fail.Load(): + http.Error(w, "down", http.StatusBadGateway) + case u.gone.Load(): + http.NotFound(w, r) + case r.URL.Path == "/big" && r.Header.Get("If-None-Match") == u.etag.Load().(string): + w.WriteHeader(http.StatusNotModified) + case r.URL.Path == "/big": + u.sent.Add(1) + body := u.body.Load().(string) + w.Header().Set("Content-Type", "application/json") + w.Header().Set("ETag", u.etag.Load().(string)) + if u.truncate.Load() { + w.Header().Set("Content-Length", fmt.Sprint(len(body))) + _, _ = io.WriteString(w, body[:len(body)/2]) + return + } + _, _ = io.WriteString(w, body) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(u.Close) + return u +} + +func streamProxy(t *testing.T, u *streamUpstream) (*Proxy, *mockStorage, http.Handler) { + t.Helper() + proxy, _, store, _ := setupTestProxy(t) + proxy.HTTPClient = u.Client() + proxy.CacheMetadata = true + proxy.MetadataTTL = time.Hour + proxy.SetMetadataRewriteCacheSize(64 << 20) + // As the server does: a cooldown.Config parses itself on first use, + // which concurrent requests would race on. + policy, err := cooldown.NewPolicy(&cooldown.Config{Default: "3d"}, nil) + if err != nil { + t.Fatal(err) + } + proxy.Cooldown = policy + return proxy, store, NewNPMHandler(proxy, "http://proxy.example", u.URL).Routes() +} + +type packumentDoc struct { + DistTags map[string]string `json:"dist-tags"` + Versions map[string]json.RawMessage `json:"versions"` + Time map[string]string `json:"time"` +} + +func decodeStreamed(t *testing.T, rec *httptest.ResponseRecorder) packumentDoc { + t.Helper() + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + body := rec.Body.Bytes() + if rec.Header().Get("Content-Encoding") == "gzip" { + zr, err := gzip.NewReader(bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + if body, err = io.ReadAll(zr); err != nil { + t.Fatal(err) + } + } + var doc packumentDoc + if err := json.Unmarshal(body, &doc); err != nil { + t.Fatalf("response is not JSON: %v", err) + } + return doc +} + +// TestNPMStreamedPackument requests a packument larger than +// metadata_max_size with cooldown on, its time map after its versions and +// its latest inside cooldown. The in-memory rewrite can't read it at all. +func TestNPMStreamedPackument(t *testing.T) { + const versions = 400 + body := bigPackument(versions) + u := newStreamUpstream(t, body) + proxy, _, handler := streamProxy(t, u) + proxy.MetadataMaxSize = int64(len(body) / 4) + + for _, header := range []http.Header{nil, {"Accept-Encoding": {"gzip"}}} { + doc := decodeStreamed(t, serve(handler, "/big", header)) + if len(doc.Versions) != versions { + t.Errorf("%d versions, want %d without 9.0.0", len(doc.Versions), versions) + } + if _, ok := doc.Versions["9.0.0"]; ok { + t.Error("9.0.0 is listed although it is inside cooldown") + } + if _, ok := doc.Time["9.0.0"]; ok { + t.Error("time still lists 9.0.0") + } + if doc.DistTags["latest"] != fmt.Sprintf("1.0.%d", versions-1) || doc.DistTags["old"] != "1.0.0" { + t.Errorf("dist-tags = %v, want latest moved to the newest version left", doc.DistTags) + } + if !strings.Contains(string(doc.Versions["1.0.0"]), `"tarball":"http://proxy.example/npm/big/-/big-1.0.0.tgz"`) { + t.Errorf("tarball not rewritten: %s", doc.Versions["1.0.0"]) + } + } + if n := u.requests.Load(); n != 1 { + t.Errorf("upstream asked %d times, want once", n) + } + + // Without the rewrite cache the in-memory path runs, which the size + // limit stops. + proxy.SetMetadataRewriteCacheSize(0) + if rec := serve(handler, "/big", nil); rec.Code == http.StatusOK { + t.Error("in-memory rewrite served a packument over metadata_max_size; the test no longer covers the stream") + } +} + +// TestNPMStreamedPackumentMatchesInMemory checks that both paths serve the +// same document and ETag. +func TestNPMStreamedPackumentMatchesInMemory(t *testing.T) { + body := bigPackument(50) + u := newStreamUpstream(t, body) + _, _, streamed := streamProxy(t, u) + inMemory, _, handler := streamProxy(t, u) + inMemory.CacheMetadata = false + + a, b := serve(streamed, "/big", nil), serve(handler, "/big", nil) + if a.Code != http.StatusOK || b.Code != http.StatusOK { + t.Fatalf("status %d and %d", a.Code, b.Code) + } + var ca, cb bytes.Buffer + if err := json.Compact(&ca, a.Body.Bytes()); err != nil { + t.Fatal(err) + } + if err := json.Compact(&cb, b.Body.Bytes()); err != nil { + t.Fatal(err) + } + if !bytes.Equal(ca.Bytes(), cb.Bytes()) { + t.Error("streamed and in-memory rewrites differ") + } +} + +// TestNPMStreamedPackumentStale serves the stored copy when upstream fails +// after the TTL, and 404 for a package upstream doesn't have. +func TestNPMStreamedPackumentStale(t *testing.T) { + u := newStreamUpstream(t, bigPackument(10)) + proxy, _, handler := streamProxy(t, u) + decodeStreamed(t, serve(handler, "/big", nil)) + + proxy.MetadataTTL = time.Nanosecond + proxy.SetMetadataRewriteCacheSize(64 << 20) // drop the cached rewrite + u.fail.Store(true) + if doc := decodeStreamed(t, serve(handler, "/big", nil)); len(doc.Versions) != 10 { + t.Errorf("stale copy has %d versions, want 10", len(doc.Versions)) + } + + u.fail.Store(false) + if rec := serve(handler, "/missing", nil); rec.Code != http.StatusNotFound { + t.Errorf("missing package: status %d, want 404", rec.Code) + } +} + +// TestNPMStreamedPackumentStaleOn404 checks that a stored copy past its TTL +// is served when upstream answers 404, as the in-memory path does. +func TestNPMStreamedPackumentStaleOn404(t *testing.T) { + u := newStreamUpstream(t, bigPackument(10)) + proxy, _, handler := streamProxy(t, u) + decodeStreamed(t, serve(handler, "/big", nil)) + + proxy.MetadataTTL = time.Nanosecond + u.gone.Store(true) + for _, size := range []int64{64 << 20, 0} { // streamed, then in memory + proxy.SetMetadataRewriteCacheSize(size) + if doc := decodeStreamed(t, serve(handler, "/big", nil)); len(doc.Versions) != 10 { + t.Errorf("rewrite cache %d: stale copy has %d versions, want 10", size, len(doc.Versions)) + } + } +} + +// TestNPMStreamedPackumentBlobMissing checks that a row whose document is +// gone from storage, within its TTL or revalidated by a 304 after it, fetches +// the document again, also when it is larger than metadata_max_size. +func TestNPMStreamedPackumentBlobMissing(t *testing.T) { + for _, tt := range []struct { + name string + ttl time.Duration + }{ + {"fresh row", time.Hour}, + {"expired row", time.Nanosecond}, + } { + t.Run(tt.name, func(t *testing.T) { + body := bigPackument(400) + u := newStreamUpstream(t, body) + proxy, store, handler := streamProxy(t, u) + proxy.MetadataMaxSize = int64(len(body) / 4) + decodeStreamed(t, serve(handler, "/big", nil)) + + path := metadataStoragePath("npm", "big") + store.mu.Lock() + delete(store.files, path) + store.mu.Unlock() + proxy.MetadataTTL = tt.ttl + proxy.SetMetadataRewriteCacheSize(64 << 20) + + if doc := decodeStreamed(t, serve(handler, "/big", nil)); len(doc.Versions) != 400 { + t.Errorf("served %d versions, want 400", len(doc.Versions)) + } + if n := u.sent.Load(); n != 2 { + t.Errorf("upstream sent the document %d times, want 2", n) + } + store.mu.Lock() + defer store.mu.Unlock() + if string(store.files[path]) != body { + t.Error("the document was not stored again") + } + }) + } +} + +// TestNPMStreamedVersionBlobMissing checks that a version request and a +// tarball download fetch a packument again whose stored copy is gone while +// its row is within the TTL. +func TestNPMStreamedVersionBlobMissing(t *testing.T) { + u := newStreamUpstream(t, "") + u.body.Store(bigPackumentAt(u.URL, 100)) + proxy, store, handler := streamProxy(t, u) + proxy.MetadataMaxSize = 1024 + fetcher := proxy.Fetcher.(*mockFetcher) + fetcher.artifact = &fetch.Artifact{Body: io.NopCloser(strings.NewReader("package"))} + path := metadataStoragePath("npm", "big") + + for _, req := range []string{"/big/1.0.5", "/big/-/big-1.0.5.tgz"} { + decodeStreamed(t, serve(handler, "/big", nil)) + store.mu.Lock() + delete(store.files, path) + store.mu.Unlock() + proxy.SetMetadataRewriteCacheSize(64 << 20) + + if rec := serve(handler, req, nil); rec.Code != http.StatusOK { + t.Errorf("%s: status %d: %.200s", req, rec.Code, rec.Body.String()) + } + store.mu.Lock() + _, ok := store.files[path] + store.mu.Unlock() + if !ok { + t.Errorf("%s: the packument was not stored again", req) + } + } +} + +// TestNPMStreamedPackumentChangedInStorage checks that a document replaced +// in storage between the two passes isn't rewritten from mixed bytes. +func TestNPMStreamedPackumentChangedInStorage(t *testing.T) { + u := newStreamUpstream(t, bigPackument(10)) + proxy, store, _ := streamProxy(t, u) + entry, err := proxy.storedMetadataDocument(context.Background(), "npm", "big", u.URL+"/big", contentTypeJSON, nil) + if err != nil { + t.Fatal(err) + } + store.mu.Lock() + store.files[entry.StoragePath] = []byte(bigPackument(11)) + store.mu.Unlock() + + r, err := proxy.openStoredMetadata(context.Background(), entry) + if err != nil { + t.Fatal(err) + } + defer func() { _ = r.Close() }() + if _, err := io.ReadAll(r); !errors.Is(err, errStoredMetadataChanged) { + t.Errorf("read error %v, want errStoredMetadataChanged", err) + } +} + +// TestNPMStreamedPackumentRevalidates checks that a 304 after the TTL keeps +// the stored document and its cached rewrite, without storing it again. +func TestNPMStreamedPackumentRevalidates(t *testing.T) { + u := newStreamUpstream(t, bigPackument(10)) + proxy, store, handler := streamProxy(t, u) + first := serve(handler, "/big", nil) + decodeStreamed(t, first) + store.mu.Lock() + stores := store.stores + store.mu.Unlock() + + proxy.MetadataTTL = time.Nanosecond + second := serve(handler, "/big", nil) + decodeStreamed(t, second) + if second.Header().Get("ETag") != first.Header().Get("ETag") { + t.Errorf("ETag changed from %s to %s", first.Header().Get("ETag"), second.Header().Get("ETag")) + } + if n := u.requests.Load(); n != 2 { + t.Errorf("upstream asked %d times, want 2", n) + } + store.mu.Lock() + defer store.mu.Unlock() + if store.stores != stores { + t.Errorf("the 304 stored the document again (%d stores, %d before)", store.stores, stores) + } +} + +// TestNPMStreamedPackumentKeepsFullCopy checks that an abbreviated packument +// from upstream, where cooldown needs the full one, doesn't replace the full +// copy in storage, which is served instead as when upstream fails. +func TestNPMStreamedPackumentKeepsFullCopy(t *testing.T) { + full := bigPackument(10) + u := newStreamUpstream(t, full) + proxy, store, handler := streamProxy(t, u) + decodeStreamed(t, serve(handler, "/big", nil)) + + proxy.MetadataTTL = time.Nanosecond + proxy.SetMetadataRewriteCacheSize(64 << 20) + u.replace(`{"name":"big","modified":"2026-01-01T00:00:00Z","dist-tags":{"latest":"1.0.0"},"versions":{"1.0.0":{}}}`, `"v2"`) + if doc := decodeStreamed(t, serve(handler, "/big", nil)); len(doc.Versions) != 10 { + t.Errorf("served %d versions, want the 10 of the full copy", len(doc.Versions)) + } + if n := u.sent.Load(); n != 2 { + t.Fatalf("upstream sent a document %d times, want 2: the abbreviated one was not fetched", n) + } + + store.mu.Lock() + defer store.mu.Unlock() + if got := string(store.files[metadataStoragePath("npm", "big")]); got != full { + t.Errorf("stored packument was replaced: %.80s", got) + } + for path := range store.files { + if strings.Contains(path, ".staged-") { + t.Errorf("staged copy %s left in storage", path) + } + } +} + +// TestNPMStreamedPackumentStorageFails checks that a packument is still +// served, read into memory, when storing it fails. +func TestNPMStreamedPackumentStorageFails(t *testing.T) { + u := newStreamUpstream(t, bigPackument(10)) + _, store, handler := streamProxy(t, u) + store.storeErr = errors.New("no space left on device") + doc := decodeStreamed(t, serve(handler, "/big", nil)) + if len(doc.Versions) != 10 || doc.DistTags["latest"] != "1.0.9" { + t.Errorf("served %d versions, latest %q; want 10 and 1.0.9", len(doc.Versions), doc.DistTags["latest"]) + } +} + +// TestNPMStreamedVersionMetadata resolves versions and dist-tags of a +// packument larger than metadata_max_size. +func TestNPMStreamedVersionMetadata(t *testing.T) { + body := bigPackument(100) + u := newStreamUpstream(t, body) + proxy, _, handler := streamProxy(t, u) + proxy.MetadataMaxSize = int64(len(body) / 4) + + for path, want := range map[string]string{ + "/big/1.0.5": "1.0.5", + "/big/latest": "1.0.99", // 9.0.0 is inside cooldown + "/big/old": "1.0.0", + } { + rec := serve(handler, path, nil) + var v struct { + Version string `json:"version"` + Dist struct { + Tarball string `json:"tarball"` + } `json:"dist"` + } + if rec.Code != http.StatusOK || json.Unmarshal(rec.Body.Bytes(), &v) != nil { + t.Errorf("%s: status %d: %.200s", path, rec.Code, rec.Body.String()) + continue + } + if v.Version != want || v.Dist.Tarball != "http://proxy.example/npm/big/-/big-"+want+".tgz" { + t.Errorf("%s = %s with tarball %s, want %s", path, v.Version, v.Dist.Tarball, want) + } + } + if rec := serve(handler, "/big/9.0.0", nil); rec.Code != http.StatusNotFound { + t.Errorf("version inside cooldown: status %d, want 404", rec.Code) + } + if n := u.requests.Load(); n != 1 { + t.Errorf("upstream asked %d times, want once", n) + } +} + +// TestNPMStreamedTarballCooldown checks the tarball download against a +// packument larger than metadata_max_size: the version's publish time and +// tarball URL are read from storage. +func TestNPMStreamedTarballCooldown(t *testing.T) { + u := newStreamUpstream(t, "") + u.body.Store(bigPackumentAt(u.URL, 100)) + proxy, _, handler := streamProxy(t, u) + proxy.MetadataMaxSize = 1024 + fetcher := proxy.Fetcher.(*mockFetcher) + fetcher.artifact = &fetch.Artifact{Body: io.NopCloser(strings.NewReader("package"))} + + if rec := serve(handler, "/big/-/big-9.0.0.tgz", nil); rec.Code != http.StatusNotFound { + t.Errorf("tarball inside cooldown: status %d, want 404", rec.Code) + } + if rec := serve(handler, "/big/-/big-1.0.5.tgz", nil); rec.Code != http.StatusOK || rec.Body.String() != "package" { + t.Errorf("tarball: status %d, body %q", rec.Code, rec.Body.String()) + } + if want := u.URL + "/big/-/big-1.0.5.tgz"; fetcher.fetchedURL != want { + t.Errorf("fetched %s, want %s from the packument", fetcher.fetchedURL, want) + } +} diff --git a/internal/handler/npm_test.go b/internal/handler/npm_test.go index 20fda4e4..9b9a6c35 100644 --- a/internal/handler/npm_test.go +++ b/internal/handler/npm_test.go @@ -561,9 +561,9 @@ func TestNPMDownloadCooldownDisabled(t *testing.T) { proxy, _, _, _ := setupTestProxy(t) h := NewNPMHandler(proxy, "http://proxy.test", "") - if inCooldown, _ := h.versionInCooldown("leftpad", testVersion100, func() ([]byte, error) { + if inCooldown, _ := h.versionInCooldown("leftpad", testVersion100, func() (npmVersionData, error) { t.Fatal("cooldown must not request metadata when disabled") - return nil, nil + return npmVersionData{}, nil }); inCooldown { t.Error("versionInCooldown = true, want false when cooldown is not configured") } diff --git a/internal/handler/rewrite_cache.go b/internal/handler/rewrite_cache.go index fd455a78..a624dbac 100644 --- a/internal/handler/rewrite_cache.go +++ b/internal/handler/rewrite_cache.go @@ -18,6 +18,7 @@ import ( "sync" "time" + "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/metrics" ) @@ -253,6 +254,18 @@ func (c *rewriteCache) rewrite(ctx context.Context, key string, in []byte, expir if c == nil { return rewriteWithETag(in, rewrite) } + return c.produce(ctx, key, expiry, func() (rewrittenMetadata, error) { + m, err := rewriteWithETag(in, rewrite) + if err == nil { + m.gzipped = gzipIfSmaller(m.plain) + } + return m, err + }) +} + +// produce is rewrite for a document that build makes with its validators +// and gzip copy. +func (c *rewriteCache) produce(ctx context.Context, key string, expiry func(start time.Time) time.Time, build func() (rewrittenMetadata, error)) (rewrittenMetadata, error) { c.mu.Lock() if e, ok := c.lookup(key, c.clock()); ok { @@ -288,10 +301,7 @@ func (c *rewriteCache) rewrite(ctx context.Context, key string, in []byte, expir c.mu.Unlock() close(f.done) }() - f.m, f.err = rewriteWithETag(in, rewrite) - if f.err == nil { - f.m.gzipped = gzipIfSmaller(f.m.plain) - } + f.m, f.err = build() return f.m, f.err } @@ -391,6 +401,60 @@ func (p *Proxy) cachedRewrite(ctx context.Context, ecosystem, proxyURL, name str return m, err } +// streamedRewrite rewrites a document stored in storage, through the +// proxy's rewrite cache, without holding the document or its rewrite in +// memory uncompressed. rewrite is given a way to open the stored document, +// as often as it needs to read it, and writes the rewritten document to w. +// The result is cached under the same key cachedRewrite uses for the +// document, and expires the same way with cooldown on. +// +// It returns errMetadataNotStored without the rewrite cache, which the +// result has to be kept in. +func (p *Proxy) streamedRewrite(ctx context.Context, ecosystem, proxyURL, name string, entry *database.MetadataCacheEntry, rewrite func(open func() (io.ReadCloser, error), w io.Writer) error) (rewrittenMetadata, error) { + if p.rewrites == nil { + return rewrittenMetadata{}, errMetadataNotStored + } + var expiry func(time.Time) time.Time + if p.cooldownEnabled() { + rec := p.watchCooldown() + defer p.unwatchCooldown(rec) + expiry = func(start time.Time) time.Time { return p.cooldownExpiry(rec, start) } + } + key := rewriteCacheKeyForDigest(ecosystem, proxyURL, name, entry.ContentDigest.String) + m, err := p.rewrites.produce(ctx, key, expiry, func() (rewrittenMetadata, error) { + // The rewrite runs to completion for the cache even when the + // caller leaves. + readCtx := context.WithoutCancel(ctx) + return gzipRewrite(func(w io.Writer) error { + return rewrite(func() (io.ReadCloser, error) { return p.openStoredMetadata(readCtx, entry) }, w) + }) + }) + if err == nil { + m.lastModified = p.metadataDate(ecosystem, proxyURL, name, m.etag) + } + return m, err +} + +// gzipRewrite runs write and keeps what it writes gzip-compressed only, with +// the ETag and length of the uncompressed output. +func gzipRewrite(write func(w io.Writer) error) (rewrittenMetadata, error) { + var buf bytes.Buffer + zw := gzip.NewWriter(&buf) + sum := sha256.New() + n := &countingWriter{w: io.Discard} + if err := write(io.MultiWriter(zw, sum, n)); err != nil { + return rewrittenMetadata{}, err + } + if err := zw.Close(); err != nil { + return rewrittenMetadata{}, err + } + return rewrittenMetadata{ + metadataValidators: metadataValidators{etag: `"` + hex.EncodeToString(sum.Sum(nil)) + `"`}, + gzipped: buf.Bytes(), + size: int(n.n), + }, nil +} + // uncachedRewrite gives a rewrite made outside the cache its validators: an // ETag of the output itself and, as cachedRewrite does, a date that replicas // agree on. @@ -501,16 +565,28 @@ func serveRewrittenMetadataAs(w http.ResponseWriter, r *http.Request, m rewritte w.WriteHeader(http.StatusNotModified) return } - body := m.gzipped - if !compressed { - body = m.content() - } w.Header().Set(headerContentType, contentType) if compressed { w.Header().Set(headerContentEncoding, "gzip") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(m.gzipped) + return + } + if m.plain == nil && m.gzipped != nil { + // Decompressed as it is sent, rather than into a copy of the + // whole document. + zr, err := gzip.NewReader(bytes.NewReader(m.gzipped)) + if err != nil { + http.Error(w, "failed to read cached metadata", http.StatusInternalServerError) + return + } + w.Header().Set(headerContentLength, strconv.Itoa(m.size)) + w.WriteHeader(http.StatusOK) + _, _ = io.Copy(w, zr) + return } w.WriteHeader(http.StatusOK) - _, _ = w.Write(body) + _, _ = w.Write(m.plain) } // ensureVary adds field to the Vary header unless it is listed already. 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 } diff --git a/internal/metadata/npm/imports_test.go b/internal/metadata/npm/imports_test.go index 0f04bc18..19f828c1 100644 --- a/internal/metadata/npm/imports_test.go +++ b/internal/metadata/npm/imports_test.go @@ -23,6 +23,7 @@ func TestPortableImports(t *testing.T) { "github.com/Masterminds/semver/v3", "github.com/git-pkgs/cooldown", "github.com/git-pkgs/proxy/internal/jsonscan", + "github.com/git-pkgs/proxy/internal/jsonstream", "github.com/git-pkgs/proxy/internal/metadata/npm", "github.com/git-pkgs/proxy/internal/packageurl", "github.com/git-pkgs/purl", diff --git a/internal/metadata/npm/stream.go b/internal/metadata/npm/stream.go new file mode 100644 index 00000000..3e86faa5 --- /dev/null +++ b/internal/metadata/npm/stream.go @@ -0,0 +1,412 @@ +package npm + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "maps" + "reflect" + "slices" + + "github.com/git-pkgs/proxy/internal/jsonstream" +) + +// ErrNotStreamable is returned by RewriteStream for a document it leaves to +// Rewrite: one without a versions object, such as a single version's +// metadata, or a rewrite that lists pushed versions. +var ErrNotStreamable = errors.New("npm metadata can't be rewritten as a stream") + +// Scan is what a first pass over a packument learns: the version names, and +// the time map and dist-tags the filters work on. It holds no version +// entries, so it stays small whatever the size of the packument. +type Scan struct { + versions []string + versionsObject bool + time any + distTags any + // occurrences counts each filtered key, so the second pass can drop the + // earlier duplicates, as Rewrite does. + occurrences map[string]int + hasModified bool +} + +// Abbreviated reports whether the scanned document is an abbreviated +// packument, as the package function Abbreviated does for a document in +// memory. +func (s *Scan) Abbreviated() bool { + return s.hasModified && s.occurrences["time"] == 0 +} + +// ScanPackument reads a packument once, keeping only what RewriteStream +// needs to decide which versions remain. Version entries are read one at a +// time and dropped. +func ScanPackument(r io.Reader) (*Scan, error) { + s := &Scan{occurrences: map[string]int{}} + dec := json.NewDecoder(r) + if err := expectDelim(dec, '{'); err != nil { + return nil, err + } + for dec.More() { + key, err := objectKey(dec) + if err != nil { + return nil, err + } + switch key { + case "versions": + s.occurrences[key]++ + s.versions, s.versionsObject, err = objectKeys(dec) + case "time": + s.occurrences[key]++ + s.time = nil + err = dec.Decode(&s.time) + case "dist-tags": + s.occurrences[key]++ + s.distTags = nil + err = dec.Decode(&s.distTags) + default: + if key == "modified" { + s.hasModified = true + } + err = skipValue(dec) + } + if err != nil { + return nil, err + } + } + if err := expectDelim(dec, '}'); err != nil { + return nil, err + } + if _, err := dec.Token(); err != io.EOF { + if err == nil { + err = errors.New("unexpected data after the top-level object") + } + return nil, err + } + return s, nil +} + +// RewriteStream is Rewrite for a packument too large to hold in memory. scan +// is the result of ScanPackument over the same bytes r reads again; the +// caller checks they are the same. The rewritten document is written to w +// with the bytes of untouched values copied as they are, and its output is +// the same as Rewrite's for the same document and options. It returns the +// versions it kept. For a document without a versions object, or options +// listing pushed versions, it returns ErrNotStreamable before writing. +func RewriteStream(packageName string, scan *Scan, r io.Reader, w io.Writer, options Options) ([]string, error) { + if !scan.versionsObject || len(options.LocalVersions) > 0 { + return nil, ErrNotStreamable + } + h := &transformer{options} + f := scan.filter(h, packageName) + sw := &streamWriter{ + transformer: h, + packageName: packageName, + scan: scan, + filtered: f, + seen: map[string]int{}, + } + if err := jsonstream.CopyObject(json.NewDecoder(r), w, sw.member); err != nil { + return nil, err + } + h.tarballsRewritten(sw.rewritten) + return slices.Collect(maps.Keys(f.versions)), nil +} + +// filtered is what cooldown and the denylist leave of a scanned packument. +type filtered struct { + versions map[string]any + timeMap, distTags map[string]any + timeChanged, tagsChanged bool +} + +// filter runs the filters Rewrite runs over what scan collected. The scan is +// left as it is, so it can be filtered again. +func (s *Scan) filter(h *transformer, packageName string) filtered { + versions := make(map[string]any, len(s.versions)) + for _, v := range s.versions { + versions[v] = nil + } + metadata := map[string]any{} + timeMap := s.objectValue(metadata, "time", s.time) + distTags := s.objectValue(metadata, "dist-tags", s.distTags) + timeEntries := len(timeMap) + originalTags := maps.Clone(distTags) + + h.applyCooldownFiltering(metadata, versions, packageName) + h.applyDenylistFiltering(metadata, versions) + + return filtered{ + versions: versions, + timeMap: timeMap, + distTags: distTags, + timeChanged: timeMap != nil && len(timeMap) != timeEntries, + tagsChanged: distTags != nil && !reflect.DeepEqual(distTags, originalTags), + } +} + +// streamWriter writes the second pass of RewriteStream from the filtered +// state the first pass left. +type streamWriter struct { + *transformer + filtered + packageName string + scan *Scan + seen map[string]int + entry bytes.Buffer + rewritten int +} + +// member writes one top-level member. Members the filters don't touch are +// left unconsumed, so CopyObject copies them as they are. +func (s *streamWriter) member(m *jsonstream.Member) error { + total, filtered := s.scan.occurrences[m.Key] + if !filtered { + return nil + } + s.seen[m.Key]++ + if s.seen[m.Key] < total { + _, err := m.Raw() // an earlier duplicate, dropped + return err + } + switch { + case m.Key == "versions": + return m.Object(s.version) + case m.Key == "time" && s.timeChanged: + return m.Object(s.time) + case m.Key == "dist-tags" && s.tagsChanged: + if _, err := m.Raw(); err != nil { + return err + } + encoded, err := json.Marshal(s.distTags) + if err != nil { + return err + } + return m.Write(encoded) + } + return nil +} + +// version writes a kept version with its tarball pointing at the proxy. +func (s *streamWriter) version(v *jsonstream.Member) error { + raw, err := v.Raw() + if err != nil { + return err + } + if _, keep := s.versions[v.Key]; !keep { + return nil + } + s.entry.Reset() + if s.writeNPMVersion(&s.entry, s.packageName, v.Key, raw) { + s.rewritten++ + } + return v.Write(s.entry.Bytes()) +} + +// time writes a time entry the filters kept. +func (s *streamWriter) time(t *jsonstream.Member) error { + if _, keep := s.timeMap[t.Key]; keep { + return t.Copy() + } + _, err := t.Raw() + return err +} + +// objectValue records value under name in metadata, as Rewrite's +// decodeObject does, and returns it when it is an object. An object is +// copied, since the filters edit it. +func (s *Scan) objectValue(metadata map[string]any, name string, value any) map[string]any { + if s.occurrences[name] == 0 { + return nil + } + object, ok := value.(map[string]any) + if !ok { + metadata[name] = value + return nil + } + object = maps.Clone(object) + metadata[name] = object + return object +} + +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 metadata: expected %v, got %v", want, tok) + } + return nil +} + +func objectKey(dec *json.Decoder) (string, error) { + tok, err := dec.Token() + if err != nil { + return "", err + } + key, ok := tok.(string) + if !ok { + return "", fmt.Errorf("npm metadata: object key is %T, not a string", tok) + } + return key, nil +} + +// objectKeys reads a value and returns its keys when it is an object. +func objectKeys(dec *json.Decoder) ([]string, bool, error) { + tok, err := dec.Token() + if err != nil { + return nil, false, err + } + delim, ok := tok.(json.Delim) + if !ok { + return nil, false, nil // a scalar, read whole by Token + } + var keys []string + for dec.More() { + if delim == '{' { + key, err := objectKey(dec) + if err != nil { + return nil, false, err + } + keys = append(keys, key) + } + if err := skipValue(dec); err != nil { + return nil, false, err + } + } + if _, err := dec.Token(); err != nil { + return nil, false, err + } + return keys, delim == '{', nil +} + +// skipValue reads the next value without keeping it. +func skipValue(dec *json.Decoder) error { + var raw json.RawMessage + return dec.Decode(&raw) +} + +// VersionStream is Version for a packument too large to hold in memory. scan +// is the result of ScanPackument over the bytes r reads again; the second +// pass only looks for the one version entry it serves. For a document +// without a versions object it returns ErrNotStreamable. +func VersionStream(packageName, specifier string, scan *Scan, r io.Reader, options Options) ([]byte, error) { + if !scan.versionsObject || len(options.LocalVersions) > 0 { + return nil, ErrNotStreamable + } + h := &transformer{options} + f := scan.filter(h, packageName) + + target := specifier + if tag, ok := f.distTags[specifier].(string); ok { + target = tag + } + if options.DeniedVersions[target] { + return nil, ErrVersionDenied + } + if _, kept := f.versions[target]; !kept { + return nil, ErrVersionNotFound + } + entry, err := LookupVersion(r, target) + if err != nil { + return nil, err + } + if entry.Entry == nil { + return nil, ErrVersionNotFound + } + var out bytes.Buffer + h.writeNPMVersion(&out, packageName, target, entry.Entry) + return out.Bytes(), nil +} + +// VersionEntry is what a packument holds for one version. +type VersionEntry struct { + // Entry is versions[version], or nil when there is none. + Entry json.RawMessage + // Published is time[version], or "" when it is missing or not a string. + Published string + // Abbreviated is set for an abbreviated packument (see Abbreviated). + Abbreviated bool +} + +// LookupVersion reads a packument once and returns what it holds for +// version, keeping no other version's entry. Of repeated keys the last one +// counts, as with JSON.parse. +func LookupVersion(r io.Reader, version string) (VersionEntry, error) { + var e VersionEntry + var hasTime, hasModified bool + dec := json.NewDecoder(r) + if err := expectDelim(dec, '{'); err != nil { + return e, err + } + for dec.More() { + key, err := objectKey(dec) + if err != nil { + return e, err + } + switch key { + case "versions": + e.Entry, err = objectMember(dec, version) + case "time": + hasTime = true + var raw json.RawMessage + raw, err = objectMember(dec, version) + e.Published = "" + if err == nil && raw != nil { + _ = json.Unmarshal(raw, &e.Published) + } + default: + hasModified = hasModified || key == "modified" + err = skipValue(dec) + } + if err != nil { + return e, err + } + } + if err := expectDelim(dec, '}'); err != nil { + return e, err + } + if _, err := dec.Token(); err != io.EOF { + if err == nil { + err = errors.New("unexpected data after the top-level object") + } + return e, err + } + e.Abbreviated = hasModified && !hasTime + return e, nil +} + +// objectMember reads a value and returns its member key when it is an object +// that has one, the last if repeated, or nil. +func objectMember(dec *json.Decoder, key string) (json.RawMessage, error) { + tok, err := dec.Token() + if err != nil { + return nil, err + } + delim, ok := tok.(json.Delim) + if !ok { + return nil, nil + } + var found json.RawMessage + for dec.More() { + var k string + if delim == '{' { + if k, err = objectKey(dec); err != nil { + return nil, err + } + } + var raw json.RawMessage + if err := dec.Decode(&raw); err != nil { + return nil, err + } + if delim == '{' && k == key { + found = raw + } + } + if _, err := dec.Token(); err != nil { + return nil, err + } + return found, nil +} diff --git a/internal/metadata/npm/stream_test.go b/internal/metadata/npm/stream_test.go new file mode 100644 index 00000000..0c8a4cee --- /dev/null +++ b/internal/metadata/npm/stream_test.go @@ -0,0 +1,225 @@ +package npm + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "slices" + "strings" + "testing" + "time" + + "github.com/git-pkgs/cooldown" +) + +// rewriteStream runs both passes of RewriteStream over body. +func rewriteStream(packageName string, body []byte, options Options) ([]byte, []string, error) { + scan, err := ScanPackument(bytes.NewReader(body)) + if err != nil { + return nil, nil, err + } + var out bytes.Buffer + kept, err := RewriteStream(packageName, scan, bytes.NewReader(body), &out, options) + return out.Bytes(), kept, err +} + +// streamCases returns packuments with the filtered members in either +// order, duplicates and values Rewrite copies verbatim, and the options to +// rewrite them with. +func streamCases() (map[string]string, map[string]Options) { + now := time.Date(2026, 10, 18, 12, 0, 0, 0, time.UTC) + ts := func(age time.Duration) string { return now.Add(-age).Format(time.RFC3339) } + version := func(v string) string { + return fmt.Sprintf(`%q:{"version":%q,"dist":{"tarball":"https://registry.npmjs.org/demo/-/demo-%s.tgz"}}`, v, v, v) + } + versions := `"versions":{` + version("1.0.0") + `,` + version("2.0.0") + `,` + version("3.0.0") + `}` + times := `"time":{"created":"` + ts(1000*time.Hour) + `","1.0.0":"` + ts(900*time.Hour) + `","2.0.0":"` + ts(800*time.Hour) + `","3.0.0":"` + ts(time.Hour) + `"}` + tags := `"dist-tags":{"latest":"3.0.0","beta":"2.0.0","next":"1.0.0"}` + + docs := map[string]string{ + "testdata": string(packument), + "tags, versions, time": `{"name":"demo",` + tags + `,` + versions + `,` + times + `}`, + "time first": `{` + times + `,"name":"demo",` + versions + `,` + tags + `}`, + "no time": `{"name":"demo",` + tags + `,` + versions + `}`, + "time null": `{"time":null,` + tags + `,` + versions + `}`, + "tags not an object": `{"dist-tags":"x",` + versions + `,` + times + `}`, + "duplicates": `{"versions":{` + version("9.0.0") + `},` + tags + `,"time":{"9.0.0":"` + ts(time.Hour) + `"},` + + versions + `,` + times + `,"dist-tags":{"latest":"3.0.0"}}`, + "verbatim values": `{"_big":12345678901234567890,"readme":"&","nested":{"a":[1, 2 ,{"b":null}]},` + versions + `}`, + "unusual entries": `{"versions":{"1.0.0":"not an object","2.0.0":{"dist":"x"},"3.0.0":{"dist":{"tarball":7}},` + + `"4.0.0":{"dist":{"tarball":"https://registry.npmjs.org/demo/-/renamed.tgz"}},"5.0.0":{"dist":{}}}}`, + "whitespace": "{ \"name\" : \"demo\" ,\n\t\"versions\" : {\n \"1.0.0\" : { \"dist\" : { \"tarball\" : \"https://registry.npmjs.org/demo/-/demo-1.0.0.tgz\" } } } }", + "empty versions": `{"name":"demo","versions":{}}`, + } + options := map[string]Options{ + "plain": {ProxyURL: "http://proxy"}, + "cooldown": {ProxyURL: "http://proxy", Cooldown: &cooldown.Config{Default: "3d"}, EvaluatedAt: now}, + "denylist": {ProxyURL: "http://proxy", DeniedVersions: map[string]bool{"2.0.0": true}}, + "both": {ProxyURL: "http://proxy", Cooldown: &cooldown.Config{Default: "3d"}, EvaluatedAt: now, + DeniedVersions: map[string]bool{"1.0.0": true}}, + } + return docs, options +} + +// TestRewriteStreamMatchesRewrite checks that the streamed rewrite writes the +// same bytes and keeps the same versions as Rewrite, over documents with the +// filtered members in either order, duplicates, and values Rewrite copies +// verbatim. +func TestRewriteStreamMatchesRewrite(t *testing.T) { + docs, options := streamCases() + for docName, doc := range docs { + for optName, opts := range options { + t.Run(docName+"/"+optName, func(t *testing.T) { + want, wantKept, err := Rewrite("demo", []byte(doc), opts) + if err != nil { + t.Fatalf("Rewrite: %v", err) + } + got, gotKept, err := rewriteStream("demo", []byte(doc), opts) + if err != nil { + t.Fatalf("RewriteStream: %v", err) + } + // Rewrite keeps the whitespace inside the members of a + // filtered time map, which the stream doesn't see. + if !bytes.Equal(compact(t, got), compact(t, want)) { + t.Errorf("RewriteStream =\n%s\nRewrite =\n%s", got, want) + } + if docName == "verbatim values" && !bytes.Contains(got, []byte(`"_big":12345678901234567890,"readme":"&","nested":{"a":[1, 2 ,{"b":null}]}`)) { + t.Errorf("values not copied verbatim: %s", got) + } + slices.Sort(gotKept) + slices.Sort(wantKept) + if !slices.Equal(gotKept, wantKept) { + t.Errorf("kept %v, Rewrite kept %v", gotKept, wantKept) + } + }) + } + } +} + +func compact(t *testing.T, doc []byte) []byte { + t.Helper() + var out bytes.Buffer + if err := json.Compact(&out, doc); err != nil { + t.Fatalf("output is not JSON: %v\n%s", err, doc) + } + return out.Bytes() +} + +func TestRewriteStreamCallbacks(t *testing.T) { + var filtered []string + var rewritten []int + options := Options{ProxyURL: "http://proxy", Cooldown: &cooldown.Config{Default: "7d"}, + EvaluatedAt: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), + OnCooldownFilter: func(version, _ string) { filtered = append(filtered, version) }, + OnTarballsRewritten: func(count int) { rewritten = append(rewritten, count) }, + } + if _, _, err := rewriteStream("@example/widgets", packument, options); err != nil { + t.Fatal(err) + } + if !slices.Equal(filtered, []string{"3.0.0"}) || !slices.Equal(rewritten, []int{2}) { + t.Errorf("filtered %v, rewritten %v; want [3.0.0] and [2]", filtered, rewritten) + } +} + +func TestRewriteStreamNotStreamable(t *testing.T) { + for name, tc := range map[string]struct { + doc string + options Options + }{ + "single version": {`{"name":"demo","version":"1.0.0","dist":{"tarball":"t"}}`, Options{}}, + "versions array": {`{"name":"demo","versions":[{"a":1}]}`, Options{}}, + "versions null": {`{"name":"demo","versions":null}`, Options{}}, + "pushed versions": {`{"versions":{}}`, Options{LocalVersions: []LocalVersion{{Version: "1.0.0", Entry: []byte(`{}`)}}}}, + } { + t.Run(name, func(t *testing.T) { + var out bytes.Buffer + scan, err := ScanPackument(strings.NewReader(tc.doc)) + if err != nil { + t.Fatal(err) + } + if _, err := RewriteStream("demo", scan, strings.NewReader(tc.doc), &out, tc.options); !errors.Is(err, ErrNotStreamable) { + t.Errorf("err = %v, want ErrNotStreamable", err) + } + if out.Len() != 0 { + t.Errorf("wrote %q before refusing", out.String()) + } + }) + } +} + +func TestScanPackument(t *testing.T) { + for _, body := range []string{`{"versions": {}`, `[1]`, `{"versions": {}} trailing`, `{"a":}`, ``} { + if _, err := ScanPackument(strings.NewReader(body)); err == nil { + t.Errorf("ScanPackument(%q): expected an error", body) + } + } + for body, want := range map[string]bool{ + `{"name":"demo","modified":"2024-01-01T00:00:00Z","versions":{}}`: true, + `{"name":"demo","modified":"2024-01-01T00:00:00Z","time":{},"versions":{}}`: false, + `{"name":"demo","versions":{}}`: false, + } { + scan, err := ScanPackument(strings.NewReader(body)) + if err != nil { + t.Fatal(err) + } + if scan.Abbreviated() != want || Abbreviated([]byte(body)) != want { + t.Errorf("Abbreviated(%s) = %v, want %v", body, scan.Abbreviated(), want) + } + } +} + +// TestVersionStreamMatchesVersion checks that a single version resolved from +// the stream is the one Version serves, with the same errors. +func TestVersionStreamMatchesVersion(t *testing.T) { + docs, options := streamCases() + specifiers := []string{"latest", "beta", "next", "old", "1.0.0", "2.0.0", "3.0.0", "9.0.0", "missing"} + for docName, doc := range docs { + scan, err := ScanPackument(strings.NewReader(doc)) + if err != nil { + t.Fatalf("%s: %v", docName, err) + } + for optName, opts := range options { + for _, spec := range specifiers { + want, wantErr := Version("demo", spec, []byte(doc), opts) + got, gotErr := VersionStream("demo", spec, scan, strings.NewReader(doc), opts) + name := docName + "/" + optName + "/" + spec + switch { + case wantErr != nil || gotErr != nil: + if !errors.Is(gotErr, ErrVersionDenied) == errors.Is(wantErr, ErrVersionDenied) || + !errors.Is(gotErr, ErrVersionNotFound) == errors.Is(wantErr, ErrVersionNotFound) || + (gotErr == nil) != (wantErr == nil) { + t.Errorf("%s: error %v, Version gave %v", name, gotErr, wantErr) + } + case !bytes.Equal(got, want): + t.Errorf("%s:\n%s\nVersion:\n%s", name, got, want) + } + } + } + } +} + +func TestLookupVersion(t *testing.T) { + doc := `{"modified":"x","versions":{"1.0.0":{"a":1}},"time":{"1.0.0":"2020-01-01T00:00:00Z","2.0.0":7},` + + `"versions":{"1.0.0":{"a":2},"2.0.0":{"b":1}}}` + for version, want := range map[string]VersionEntry{ + "1.0.0": {Entry: json.RawMessage(`{"a":2}`), Published: "2020-01-01T00:00:00Z"}, + "2.0.0": {Entry: json.RawMessage(`{"b":1}`)}, + "3.0.0": {}, + } { + got, err := LookupVersion(strings.NewReader(doc), version) + if err != nil { + t.Fatal(err) + } + if string(got.Entry) != string(want.Entry) || got.Published != want.Published || got.Abbreviated { + t.Errorf("LookupVersion(%s) = %+v, want %+v", version, got, want) + } + } + got, err := LookupVersion(strings.NewReader(`{"modified":"x","versions":{}}`), "1.0.0") + if err != nil || !got.Abbreviated { + t.Errorf("abbreviated packument: %+v, %v", got, err) + } + if _, err := LookupVersion(strings.NewReader(`{"versions":{}`), "1.0.0"); err == nil { + t.Error("expected an error for truncated JSON") + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 69363ca8..8ca80d16 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -262,10 +262,7 @@ func (s *Server) serve(listener net.Listener) error { if err != nil { return fmt.Errorf("configuring scanners: %w", err) } - proxy.CacheMetadata = s.cfg.CacheMetadata - proxy.MetadataTTL = s.cfg.ParseMetadataTTL() - proxy.MetadataMaxSize = s.cfg.ParseMetadataMaxSize() - proxy.SetMetadataRewriteCacheSize(s.cfg.ParseMetadataRewriteCacheSize()) + configureMetadata(proxy, s.cfg) proxy.GradleReadOnly = s.cfg.Gradle.BuildCache.ReadOnly proxy.HostedEnabled = s.cfg.Hosted.Enabled proxy.HostedNPMScopes = s.cfg.Hosted.NPMScopes @@ -616,6 +613,15 @@ func (s *Server) newContainerHandler(proxy *handler.Proxy) *handler.ContainerHan return containerHandler } +// configureMetadata applies the metadata cache settings to proxy. +func configureMetadata(proxy *handler.Proxy, cfg *config.Config) { + proxy.CacheMetadata = cfg.CacheMetadata + proxy.MetadataTTL = cfg.ParseMetadataTTL() + proxy.MetadataMaxSize = cfg.ParseMetadataMaxSize() + proxy.MetadataStreamMaxSize = cfg.ParseMetadataStreamMaxSize() + proxy.SetMetadataRewriteCacheSize(cfg.ParseMetadataRewriteCacheSize()) +} + func configureScanning(proxy *handler.Proxy, cfg config.ScanningConfig, baseURL string, logger *slog.Logger) (*scanner.Group, error) { scanGroup, err := scanner.NewGroup(cfg, logger) if err != nil {