Mirror packages from the configured upstreams

docs/configuration.md

@@ -1385,6 +1385,8 @@ Pushed archives are not touched: a whole-version delete skips them, and naming o
 
 The `proxy mirror` command pre-populates the cache from various sources. It accepts the same storage and database flags as `serve`.
 
+Packages are downloaded from the configured upstreams: npm, cargo, gem, golang, hex, pub, maven and nuget from `upstream.npm`, `upstream.cargo_download`, `upstream.gem`, `upstream.go`, `upstream.hex`, `upstream.pub`, `upstream.maven` and `upstream.nuget`. npm packages in `upstream.npm_routes` come from their route and are cached under the route's keys, so clients get them from the cache as they would after an install. Mirror API jobs do the same.
+
 | Flag | Default | Description |
 |------|---------|-------------|
 | `--sbom` | | Path to CycloneDX or SPDX SBOM file |

internal/handler/handler.go

@@ -152,10 +152,14 @@ func (p *Proxy) ReadMetadata(r io.Reader) ([]byte, error) {
 
 // Proxy provides shared functionality for protocol handlers.
 type Proxy struct {
-	DB                  *database.DB
-	Storage             storage.Storage
-	Fetcher             fetch.FetcherInterface
-	Resolver            *fetch.Resolver
+	DB       *database.DB
+	Storage  storage.Storage
+	Fetcher  fetch.FetcherInterface
+	Resolver *fetch.Resolver
+	// RouteFor returns how to fetch a package that doesn't come from the
+	// ecosystem's default upstream, such as an npm package in
+	// upstream.npm_routes, or nil to use Resolver.
+	RouteFor            func(ecosystem, name string) *ArtifactRoute
 	Logger              *slog.Logger
 	Cooldown            CooldownPolicy
 	Denylist            *denylist.Policy
@@ -291,6 +295,9 @@ func (p *Proxy) GetOrFetchArtifact(ctx context.Context, ecosystem, name, version
 		if filename == "" {
 			return nil, errors.New("resolved artifact has no filename")
 		}
+		if route := p.routeFor(ecosystem, name); route != nil {
+			filename = route.FilenamePrefix + filename
+		}
 	}
 	if p.StreamArtifacts {
 		if info == nil {
@@ -521,8 +528,29 @@ func (p *Proxy) rejectUnusableCacheRecord(artifact *database.CachedArtifact, ver
 	}
 }
 
+// ArtifactRoute fetches a package from an upstream other than its
+// ecosystem's default one.
+type ArtifactRoute struct {
+	// Resolver finds the package's download URLs on that upstream.
+	Resolver *fetch.Resolver
+	// FilenamePrefix qualifies the artifact filename the way the handler
+	// does for the route, so clients are served what was fetched.
+	FilenamePrefix string
+}
+
+func (p *Proxy) routeFor(ecosystem, name string) *ArtifactRoute {
+	if p.RouteFor == nil {
+		return nil
+	}
+	return p.RouteFor(ecosystem, name)
+}
+
 func (p *Proxy) resolveArtifact(ctx context.Context, ecosystem, name, version string) (*fetch.ArtifactInfo, error) {
-	info, err := p.Resolver.Resolve(ctx, ecosystem, name, version)
+	resolver := p.Resolver
+	if route := p.routeFor(ecosystem, name); route != nil {
+		resolver = route.Resolver
+	}
+	info, err := resolver.Resolve(ctx, ecosystem, name, version)
 	if err != nil {
 		if errors.Is(err, fetch.ErrNotFound) {
 			return nil, ErrUpstreamNotFound

internal/handler/package_routes.go

@@ -30,11 +30,10 @@ func newPackageRoutes(patterns map[string]string) packageRoutes {
 	routes := make([]packageRoute, 0, len(patterns))
 	for pattern, upstreamURL := range patterns {
 		upstreamURL = strings.TrimSuffix(upstreamURL, "/")
-		digest := sha256.Sum256([]byte(upstreamURL))
 		routes = append(routes, packageRoute{
 			pattern: strings.ToLower(pattern),
 			url:     upstreamURL,
-			id:      hex.EncodeToString(digest[:])[:routeIDLength],
+			id:      routeID(upstreamURL),
 		})
 	}
 	sort.Slice(routes, func(i, j int) bool {
@@ -46,6 +45,18 @@ func newPackageRoutes(patterns map[string]string) packageRoutes {
 	return packageRoutes{routes: routes}
 }
 
+// routeID identifies a route's upstream in cache keys and filenames.
+func routeID(upstreamURL string) string {
+	digest := sha256.Sum256([]byte(strings.TrimSuffix(upstreamURL, "/")))
+	return hex.EncodeToString(digest[:])[:routeIDLength]
+}
+
+// RoutedFilenamePrefix returns what the npm and Composer handlers put in
+// front of the artifact filenames of a package routed to routeURL.
+func RoutedFilenamePrefix(routeURL string) string {
+	return routeID(routeURL) + "/"
+}
+
 // match returns the route for a package name, if any pattern matches it.
 func (r packageRoutes) match(packageName string) (packageRoute, bool) {
 	name := strings.ToLower(packageName)
@@ -72,3 +83,10 @@ func (r packageRoutes) cacheFilename(packageName, filename string) string {
 	}
 	return filename
 }
+
+// RoutedPackageURL returns the upstream URL a route in routes sends
+// packageName to, as the npm and Composer handlers pick it.
+func RoutedPackageURL(routes map[string]string, packageName string) (string, bool) {
+	route, ok := newPackageRoutes(routes).match(packageName)
+	return route.url, ok
+}

internal/server/mirror_fetch_test.go

@@ -0,0 +1,135 @@
+package server
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"github.com/git-pkgs/proxy/internal/config"
+	"github.com/git-pkgs/proxy/internal/database"
+	"github.com/git-pkgs/proxy/internal/handler"
+	"github.com/git-pkgs/proxy/internal/mirror"
+	"github.com/git-pkgs/proxy/internal/storage"
+)
+
+// newTarballUpstream serves body for path, compared case-insensitively since
+// scoped npm names arrive with an escaped slash, and counts the requests.
+func newTarballUpstream(t *testing.T, path, body string) (*httptest.Server, *atomic.Int32) {
+	t.Helper()
+	var hits atomic.Int32
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !strings.EqualFold(r.URL.EscapedPath(), path) && !strings.EqualFold(r.URL.Path, path) {
+			http.NotFound(w, r)
+			return
+		}
+		hits.Add(1)
+		_, _ = io.WriteString(w, body)
+	}))
+	t.Cleanup(upstream.Close)
+	return upstream, &hits
+}
+
+// mirrorWithConfig mirrors purls with the proxy proxy mirror builds for cfg.
+func mirrorWithConfig(t *testing.T, cfg *config.Config, purls ...string) (*mirror.Progress, *handler.Proxy) {
+	t.Helper()
+	cfg.Upstream.AllowLoopback = true
+	db, err := database.Create(t.TempDir() + "/test.db")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.MigrateSchema(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	store, err := storage.OpenBucket(context.Background(), "file://"+t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	proxy, _ := NewMirrorProxy(cfg, db, store, logger)
+	progress, err := mirror.New(proxy, db, store, logger, 1).Run(context.Background(), &mirror.PURLSource{PURLs: purls})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	return progress, proxy
+}
+
+func TestMirrorFetchesFromConfiguredUpstream(t *testing.T) {
+	upstream, hits := newTarballUpstream(t, "/left-pad/-/left-pad-1.3.0.tgz", "tarball")
+	cfg := config.Default()
+	cfg.Upstream.NPM = upstream.URL
+
+	progress, _ := mirrorWithConfig(t, cfg, "pkg:npm/left-pad@1.3.0")
+
+	if progress.Completed != 1 || hits.Load() != 1 {
+		t.Errorf("completed = %d, upstream hits = %d, errors = %v; want the tarball from upstream.npm", progress.Completed, hits.Load(), progress.Errors)
+	}
+}
+
+func TestMirrorFetchesRoutedNPMPackagesFromTheirRoute(t *testing.T) {
+	const tarballPath = "/@acme/widget/-/widget-1.0.0.tgz"
+	public, publicHits := newTarballUpstream(t, tarballPath, "public")
+	private, privateHits := newTarballUpstream(t, tarballPath, "private")
+	cfg := config.Default()
+	cfg.Upstream.NPM = public.URL
+	cfg.Upstream.NPMRoutes = map[string]string{"@acme/*": private.URL}
+
+	progress, proxy := mirrorWithConfig(t, cfg, "pkg:npm/%40acme/widget@1.0.0")
+
+	if progress.Completed != 1 || privateHits.Load() != 1 {
+		t.Errorf("completed = %d, route hits = %d, errors = %v; want the tarball from the route", progress.Completed, privateHits.Load(), progress.Errors)
+	}
+	if publicHits.Load() != 0 {
+		t.Error("a routed package was mirrored from upstream.npm")
+	}
+
+	// A client gets the mirrored tarball from the cache once the route is down.
+	private.Close()
+	npm := handler.NewNPMHandler(proxy, "http://proxy.test", cfg.Upstream.NPM).WithPackageRoutes(cfg.Upstream.NPMRoutes)
+	rec := httptest.NewRecorder()
+	npm.Routes().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, tarballPath, nil))
+	if rec.Code != http.StatusOK || rec.Body.String() != "private" {
+		t.Errorf("tarball from cache: status = %d, body = %q, want 200 and the mirrored tarball", rec.Code, rec.Body)
+	}
+}
+
+func TestMirrorResolverBaseURLs(t *testing.T) {
+	u := config.Default().Upstream
+	u.NPM = "https://npm.example.com/"
+	u.CargoDownload = "https://crates.example.com/crates"
+	u.Gem = "https://gems.example.com"
+	u.Go = "https://goproxy.example.com"
+	u.Hex = "https://hex.example.com"
+	u.Pub = "https://pub.example.com"
+	u.Maven = "https://maven.example.com/maven2"
+	u.NuGet = "https://nuget.example.com"
+
+	resolver := newMirrorResolver(u)
+	tests := []struct {
+		ecosystem, name, version, want string
+	}{
+		{"npm", "left-pad", "1.3.0", "https://npm.example.com/left-pad/-/left-pad-1.3.0.tgz"},
+		{"cargo", "serde", "1.0.0", "https://crates.example.com/crates/serde/serde-1.0.0.crate"},
+		{"gem", "rake", "13.2.1", "https://gems.example.com/downloads/rake-13.2.1.gem"},
+		{"golang", "golang.org/x/mod", "v0.20.0", "https://goproxy.example.com/golang.org/x/mod/@v/v0.20.0.zip"},
+		{"hex", "jason", "1.4.4", "https://hex.example.com/tarballs/jason-1.4.4.tar"},
+		{"pub", "http", "1.2.2", "https://pub.example.com/packages/http/versions/1.2.2.tar.gz"},
+		{"maven", "org.example:lib", "1.0", "https://maven.example.com/maven2/org/example/lib/1.0/lib-1.0.jar"},
+		{"nuget", "Newtonsoft.Json", "13.0.3", "https://nuget.example.com/v3-flatcontainer/newtonsoft.json/13.0.3/newtonsoft.json.13.0.3.nupkg"},
+	}
+	for _, tt := range tests {
+		info, err := resolver.Resolve(context.Background(), tt.ecosystem, tt.name, tt.version)
+		if err != nil {
+			t.Errorf("%s: Resolve() error = %v", tt.ecosystem, err)
+			continue
+		}
+		if info.URL != tt.want {
+			t.Errorf("%s: URL = %q, want %q", tt.ecosystem, info.URL, tt.want)
+		}
+	}
+}

internal/server/mirror_resolver.go

@@ -0,0 +1,52 @@
+package server
+
+import (
+	"github.com/git-pkgs/proxy/internal/config"
+	"github.com/git-pkgs/proxy/internal/handler"
+	"github.com/git-pkgs/registries/fetch"
+)
+
+// newMirrorResolver builds the resolver proxy mirror and mirror jobs find
+// download URLs with, using the configured upstreams. Handlers serving
+// clients don't use it; they take the URLs from upstream metadata.
+func newMirrorResolver(u config.UpstreamConfig) *fetch.Resolver {
+	return fetch.NewResolver(fetch.WithBaseURLs(map[string]string{
+		"npm":    u.NPM,
+		"cargo":  u.CargoDownload,
+		"gem":    u.Gem,
+		"golang": u.Go,
+		"hex":    u.Hex,
+		"pub":    u.Pub,
+		"maven":  u.Maven,
+		"nuget":  u.NuGet,
+	}))
+}
+
+// configureMirrorResolvers points proxy's resolver at the configured
+// upstreams, and npm packages in upstream.npm_routes at their route, so a
+// public package can't stand in for a routed one.
+func configureMirrorResolvers(proxy *handler.Proxy, u config.UpstreamConfig) {
+	proxy.Resolver = newMirrorResolver(u)
+	if len(u.NPMRoutes) == 0 {
+		return
+	}
+	routed := make(map[string]*handler.ArtifactRoute, len(u.NPMRoutes))
+	for _, routeURL := range u.NPMRoutes {
+		routeUpstream := u
+		routeUpstream.NPM = routeURL
+		routed[routeURL] = &handler.ArtifactRoute{
+			Resolver:       newMirrorResolver(routeUpstream),
+			FilenamePrefix: handler.RoutedFilenamePrefix(routeURL),
+		}
+	}
+	proxy.RouteFor = func(ecosystem, name string) *handler.ArtifactRoute {
+		if ecosystem != "npm" {
+			return nil
+		}
+		routeURL, ok := handler.RoutedPackageURL(u.NPMRoutes, name)
+		if !ok {
+			return nil
+		}
+		return routed[routeURL]
+	}
+}

internal/server/server.go

@@ -234,13 +234,13 @@ func (s *Server) serve(listener net.Listener) error {
 	baseFetcher := fetch.NewFetcher(fetch.WithHTTPClient(artifactClient), fetch.WithMaxArtifactSize(s.cfg.ParseArtifactMaxSize()))
 	fetcher := fetch.NewCircuitBreakerFetcher(baseFetcher)
 	s.breakers = newBreakerMonitor(fetcher, s.logger)
-	resolver := fetch.NewResolver()
 	cd := &cooldown.Config{
 		Default:    s.cfg.Cooldown.Default,
 		Ecosystems: s.cfg.Cooldown.Ecosystems,
 		Packages:   s.cfg.Cooldown.NormalizedPackages(),
 	}
-	proxy := handler.NewProxy(s.db, s.storage, fetcher, resolver, s.logger)
+	proxy := handler.NewProxy(s.db, s.storage, fetcher, nil, s.logger)
+	configureMirrorResolvers(proxy, s.cfg.Upstream)
 	proxy.HTTPClient = metadataClient
 	proxy.RateLimits = s.rateLimits
 	cooldownPolicy, err := cooldown.NewPolicy(cd, s.cfg.Cooldown.PackagePatterns)
@@ -575,7 +575,8 @@ func NewMirrorProxy(cfg *config.Config, db *database.DB, store storage.Storage,
 	s := &Server{cfg: cfg, db: db, storage: store, logger: logger, ecr: newECRTokens(logger)}
 	metadataClient, artifactClient := s.newUpstreamClients(0)
 	fetcher := fetch.NewFetcher(fetch.WithHTTPClient(artifactClient), fetch.WithMaxArtifactSize(cfg.ParseArtifactMaxSize()))
-	proxy := handler.NewProxy(db, store, fetcher, fetch.NewResolver(), logger)
+	proxy := handler.NewProxy(db, store, fetcher, nil, logger)
+	configureMirrorResolvers(proxy, cfg.Upstream)
 	proxy.HTTPClient = metadataClient
 	proxy.RateLimits = s.rateLimits
 	return proxy, s.newContainerHandler(proxy)