From 055540a540614aa7d01dc550ca5cab451211b77b Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Sun, 11 Oct 2026 23:59:29 +0200 Subject: [PATCH 1/3] Manifests are now cached under a key that ignores the client's Accept header: a digest names one document, and a tag is shared by all clients that accept an index type, so Docker, containerd and Podman read the same entry and a pre-fetched image is found offline. Clients without index support keep their per-Accept entry, and the old keys are still written and read as fallback. --- internal/handler/container_manifest.go | 95 +++++++--- .../handler/container_manifest_key_test.go | 165 ++++++++++++++++++ internal/handler/container_test.go | 10 +- 3 files changed, 240 insertions(+), 30 deletions(-) create mode 100644 internal/handler/container_manifest_key_test.go diff --git a/internal/handler/container_manifest.go b/internal/handler/container_manifest.go index 01df3afb..4ff750ed 100644 --- a/internal/handler/container_manifest.go +++ b/internal/handler/container_manifest.go @@ -8,6 +8,7 @@ import ( "mime" "net/http" "regexp" + "slices" "sort" "strconv" "strings" @@ -23,6 +24,14 @@ const ( containerAcceptExactSpecificity ) +const ( + // containerManifestIndexCacheClass replaces the Accept header in tag cache + // keys for clients that accept an index type. + containerManifestIndexCacheClass = "index" + ociImageIndexMediaType = "application/vnd.oci.image.index.v1+json" + dockerManifestListMediaType = "application/vnd.docker.distribution.manifest.list.v2+json" +) + var manifestDigestReferencePattern = regexp.MustCompile(`^[a-z0-9]+:[a-f0-9]+$`) type cachedContainerManifest struct { @@ -38,8 +47,7 @@ type cachedContainerManifest struct { func (h *ContainerHandler) serveManifest(w http.ResponseWriter, r *http.Request, registryURL, name, reference string) { accept := containerManifestAccept(r) cacheAccept := normalizeContainerManifestAccept(accept) - cacheKey := h.containerManifestCacheKey(registryURL, name, reference, cacheAccept) - cached := h.loadContainerManifestForAccept(r.Context(), registryURL, name, reference, accept, cacheKey) + cached := h.loadContainerManifestForAccept(r.Context(), registryURL, name, reference, accept, cacheAccept) immutable := manifestDigestReferencePattern.MatchString(reference) if cached != nil && (immutable || h.containerManifestFresh(cached)) { @@ -140,7 +148,44 @@ func (h *ContainerHandler) containerManifestCacheKey(registryURL, name, referenc return hex.EncodeToString(sum[:]) } -func (h *ContainerHandler) loadContainerManifestForAccept(ctx context.Context, registryURL, name, reference, accept, cacheKey string) *cachedContainerManifest { +// containerManifestSharedCacheKey is the key clients with different Accept +// headers share. A digest names exactly one document. A tag is stored once for +// all clients that accept an index type, whatever upstream returned for it, so +// Docker, containerd and Podman read the same entry. Clients without index +// support keep a key per normalized Accept header, because registries may +// convert an index into a single platform manifest for them. +func (h *ContainerHandler) containerManifestSharedCacheKey(registryURL, name, reference, cacheAccept string) string { + if manifestDigestReferencePattern.MatchString(reference) { + identity := strings.Join([]string{registryURL, name, reference}, "\x00") + sum := sha256.Sum256([]byte(identity)) + return hex.EncodeToString(sum[:]) + } + if containerManifestAcceptsIndex(cacheAccept) { + return h.containerManifestCacheKey(registryURL, name, reference, containerManifestIndexCacheClass) + } + return h.containerManifestCacheKey(registryURL, name, reference, cacheAccept) +} + +func containerManifestAcceptsIndex(accept string) bool { + return containerManifestAccepts(accept, ociImageIndexMediaType) || containerManifestAccepts(accept, dockerManifestListMediaType) +} + +// containerManifestLegacyCacheKeys are the per-Accept keys older versions +// used. They are still written so a rollback finds the cache, and read when +// the shared key has no entry yet. +func (h *ContainerHandler) containerManifestLegacyCacheKeys(registryURL, name, reference, accept, cacheAccept, sharedKey string) []string { + var keys []string + for _, variant := range []string{cacheAccept, accept} { + key := h.containerManifestCacheKey(registryURL, name, reference, variant) + if key != sharedKey && !slices.Contains(keys, key) { + keys = append(keys, key) + } + } + return keys +} + +func (h *ContainerHandler) loadContainerManifestForAccept(ctx context.Context, registryURL, name, reference, accept, cacheAccept string) *cachedContainerManifest { + cacheKey := h.containerManifestSharedCacheKey(registryURL, name, reference, cacheAccept) cached, err := h.loadContainerManifest(ctx, cacheKey) if err != nil { h.proxy.Logger.Warn("failed to read cached container manifest", "error", err) @@ -153,36 +198,36 @@ func (h *ContainerHandler) loadContainerManifestForAccept(ctx context.Context, r return nil } - legacyCacheKey := h.containerManifestCacheKey(registryURL, name, reference, accept) - if legacyCacheKey == cacheKey { - return nil - } - cached, err = h.loadContainerManifest(ctx, legacyCacheKey) - if err != nil { - h.proxy.Logger.Warn("failed to read legacy cached container manifest", "error", err) - return nil - } - if cached == nil || !containerManifestCacheCompatible(accept, cached) { - return nil - } - if err := h.storeContainerManifest(ctx, cacheKey, cached); err != nil { - h.proxy.Logger.Warn("failed to migrate cached container manifest", "error", err) + for _, legacyCacheKey := range h.containerManifestLegacyCacheKeys(registryURL, name, reference, accept, cacheAccept, cacheKey) { + cached, err = h.loadContainerManifest(ctx, legacyCacheKey) + if err != nil { + h.proxy.Logger.Warn("failed to read legacy cached container manifest", "error", err) + return nil + } + if cached == nil { + continue + } + if !containerManifestCacheCompatible(accept, cached) { + return nil + } + if err := h.storeContainerManifest(ctx, cacheKey, cached); err != nil { + h.proxy.Logger.Warn("failed to migrate cached container manifest", "error", err) + } + return cached } - return cached + return nil } func (h *ContainerHandler) storeContainerManifestForAccept(ctx context.Context, registryURL, name, reference, accept, cacheAccept string, manifest *cachedContainerManifest) { - cacheKey := h.containerManifestCacheKey(registryURL, name, reference, cacheAccept) + cacheKey := h.containerManifestSharedCacheKey(registryURL, name, reference, cacheAccept) if err := h.storeContainerManifest(ctx, cacheKey, manifest); err != nil { h.proxy.Logger.Warn("failed to cache container manifest", "error", err) } - legacyCacheKey := h.containerManifestCacheKey(registryURL, name, reference, accept) - if legacyCacheKey == cacheKey { - return - } - if err := h.storeContainerManifest(ctx, legacyCacheKey, manifest); err != nil { - h.proxy.Logger.Warn("failed to cache legacy container manifest", "error", err) + for _, legacyCacheKey := range h.containerManifestLegacyCacheKeys(registryURL, name, reference, accept, cacheAccept, cacheKey) { + if err := h.storeContainerManifest(ctx, legacyCacheKey, manifest); err != nil { + h.proxy.Logger.Warn("failed to cache legacy container manifest", "error", err) + } } } diff --git a/internal/handler/container_manifest_key_test.go b/internal/handler/container_manifest_key_test.go new file mode 100644 index 00000000..e339ced0 --- /dev/null +++ b/internal/handler/container_manifest_key_test.go @@ -0,0 +1,165 @@ +package handler + +import ( + "io" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" +) + +const ( + dockerManifestAccept = "application/vnd.docker.distribution.manifest.v2+json, " + + "application/vnd.docker.distribution.manifest.list.v2+json, " + + "application/vnd.oci.image.index.v1+json, application/vnd.oci.image.manifest.v1+json" + containerdManifestAccept = "application/vnd.docker.distribution.manifest.v2+json, " + + "application/vnd.docker.distribution.manifest.list.v2+json, " + + "application/vnd.oci.image.manifest.v1+json, application/vnd.oci.image.index.v1+json, */*" + ociManifestMediaType = "application/vnd.oci.image.manifest.v1+json" +) + +// fakeManifestRegistry serves one document per manifest path and fails with +// 503 once it is switched off. +type fakeManifestRegistry struct { + server *httptest.Server + offline atomic.Bool + docs map[string]fakeManifest +} + +type fakeManifest struct { + contentType string + body string +} + +func newFakeManifestRegistry(t *testing.T) *fakeManifestRegistry { + t.Helper() + registry := &fakeManifestRegistry{docs: make(map[string]fakeManifest)} + registry.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if registry.offline.Load() { + http.Error(w, "offline", http.StatusServiceUnavailable) + return + } + doc, ok := registry.docs[r.URL.Path] + if !ok { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", doc.contentType) + w.Header().Set("Docker-Content-Digest", sha256Digest([]byte(doc.body))) + _, _ = io.WriteString(w, doc.body) + })) + t.Cleanup(registry.server.Close) + return registry +} + +func (r *fakeManifestRegistry) set(path, contentType, body string) { + r.docs[path] = fakeManifest{contentType: contentType, body: body} +} + +func newManifestKeyTestHandler(t *testing.T, registry *fakeManifestRegistry) *ContainerHandler { + t.Helper() + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = registry.server.Client() + // A zero TTL makes every tag request revalidate, so an offline upstream + // exercises the stale path a real offline pull takes. + proxy.MetadataTTL = 0 + return &ContainerHandler{proxy: proxy, registryURL: registry.server.URL, proxyURL: "http://localhost:8080"} +} + +func getManifest(t *testing.T, h *ContainerHandler, path, accept string) *httptest.ResponseRecorder { + t.Helper() + request := httptest.NewRequest(http.MethodGet, path, nil) + request.Header.Set("Accept", accept) + response := httptest.NewRecorder() + h.Routes().ServeHTTP(response, request) + return response +} + +func TestContainerHandler_SingleArchTagSharedAcrossClients(t *testing.T) { + manifest := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json"}` + registry := newFakeManifestRegistry(t) + registry.set("/v2/library/app/manifests/1.0", ociManifestMediaType, manifest) + h := newManifestKeyTestHandler(t, registry) + + if response := getManifest(t, h, "/library/app/manifests/1.0", dockerManifestAccept); response.Code != http.StatusOK { + t.Fatalf("docker pull status = %d: %s", response.Code, response.Body.String()) + } + registry.offline.Store(true) + + response := getManifest(t, h, "/library/app/manifests/1.0", containerdManifestAccept) + if response.Code != http.StatusOK { + t.Fatalf("offline containerd pull status = %d, want 200: %s", response.Code, response.Body.String()) + } + if response.Body.String() != manifest { + t.Errorf("body = %q, want %q", response.Body.String(), manifest) + } +} + +func TestContainerHandler_DigestManifestSharedAcrossClients(t *testing.T) { + manifest := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json","config":{}}` + digest := sha256Digest([]byte(manifest)) + registry := newFakeManifestRegistry(t) + registry.set("/v2/library/app/manifests/"+digest, ociManifestMediaType, manifest) + h := newManifestKeyTestHandler(t, registry) + + if response := getManifest(t, h, "/library/app/manifests/"+digest, dockerManifestAccept); response.Code != http.StatusOK { + t.Fatalf("docker pull status = %d: %s", response.Code, response.Body.String()) + } + registry.offline.Store(true) + + response := getManifest(t, h, "/library/app/manifests/"+digest, ociManifestMediaType) + if response.Code != http.StatusOK { + t.Fatalf("offline pull by digest status = %d, want 200: %s", response.Code, response.Body.String()) + } +} + +func TestContainerHandler_RepushedTagServesLatestDocument(t *testing.T) { + index := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.index.v1+json","manifests":[]}` + manifest := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json"}` + registry := newFakeManifestRegistry(t) + registry.set("/v2/library/app/manifests/latest", ociImageIndexMediaType, index) + h := newManifestKeyTestHandler(t, registry) + + if response := getManifest(t, h, "/library/app/manifests/latest", dockerManifestAccept); response.Body.String() != index { + t.Fatalf("first pull body = %q, want index", response.Body.String()) + } + registry.set("/v2/library/app/manifests/latest", ociManifestMediaType, manifest) + if response := getManifest(t, h, "/library/app/manifests/latest", containerdManifestAccept); response.Body.String() != manifest { + t.Fatalf("pull after re-push body = %q, want manifest", response.Body.String()) + } + registry.offline.Store(true) + + response := getManifest(t, h, "/library/app/manifests/latest", dockerManifestAccept) + if response.Code != http.StatusOK { + t.Fatalf("offline pull status = %d, want 200: %s", response.Code, response.Body.String()) + } + if response.Body.String() != manifest { + t.Errorf("offline body = %q, want the re-pushed manifest", response.Body.String()) + } +} + +func TestContainerHandler_IndexlessClientKeepsOwnTagEntry(t *testing.T) { + index := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.index.v1+json","manifests":[]}` + converted := `{"schemaVersion":2,"mediaType":"application/vnd.docker.distribution.manifest.v2+json"}` + const schema2 = "application/vnd.docker.distribution.manifest.v2+json" + registry := newFakeManifestRegistry(t) + registry.set("/v2/library/app/manifests/latest", schema2, converted) + h := newManifestKeyTestHandler(t, registry) + + // A client without index support gets the converted platform manifest. + if response := getManifest(t, h, "/library/app/manifests/latest", schema2); response.Body.String() != converted { + t.Fatalf("index-less pull body = %q", response.Body.String()) + } + registry.set("/v2/library/app/manifests/latest", ociImageIndexMediaType, index) + if response := getManifest(t, h, "/library/app/manifests/latest", dockerManifestAccept); response.Body.String() != index { + t.Fatalf("index-capable pull body = %q, want the index", response.Body.String()) + } + registry.offline.Store(true) + + if response := getManifest(t, h, "/library/app/manifests/latest", containerdManifestAccept); response.Body.String() != index { + t.Errorf("offline index-capable body = %q, want the index", response.Body.String()) + } + if response := getManifest(t, h, "/library/app/manifests/latest", schema2); response.Body.String() != converted { + t.Errorf("offline index-less body = %q, want the converted manifest", response.Body.String()) + } +} diff --git a/internal/handler/container_test.go b/internal/handler/container_test.go index bd237cf7..bd165c2e 100644 --- a/internal/handler/container_test.go +++ b/internal/handler/container_test.go @@ -889,7 +889,7 @@ func TestContainerHandler_ManifestDigestMismatchIsNotCached(t *testing.T) { if !strings.Contains(w.Body.String(), "DIGEST_INVALID") { t.Errorf("body = %q, want DIGEST_INVALID", w.Body.String()) } - cacheKey := h.containerManifestCacheKey(upstream.URL, "homebrew/core/jq", digest, containerManifestAccept(req)) + cacheKey := h.containerManifestSharedCacheKey(upstream.URL, "homebrew/core/jq", digest, normalizeContainerManifestAccept(containerManifestAccept(req))) entry, err := db.GetMetadataCache(containerManifestCacheEcosystem, cacheKey) if err != nil { t.Fatalf("checking manifest cache: %v", err) @@ -951,7 +951,7 @@ func TestContainerHandler_ManifestTagWithInvalidDigestIsNotAliased(t *testing.T) if w.Code != http.StatusBadGateway { t.Fatalf("status = %d, want %d; body: %s", w.Code, http.StatusBadGateway, w.Body.String()) } - cacheKey := h.containerManifestCacheKey(upstream.URL, "homebrew/core/jq", invalidDigest, containerManifestAccept(req)) + cacheKey := h.containerManifestSharedCacheKey(upstream.URL, "homebrew/core/jq", invalidDigest, normalizeContainerManifestAccept(containerManifestAccept(req))) entry, err := db.GetMetadataCache(containerManifestCacheEcosystem, cacheKey) if err != nil { t.Fatalf("checking manifest cache: %v", err) @@ -1075,9 +1075,9 @@ func TestContainerHandler_ManifestMigratesLegacyAcceptCacheKey(t *testing.T) { accept := "application/vnd.oci.image.index.v1+json;q=1, application/vnd.oci.image.manifest.v1+json;q=1" cacheAccept := normalizeContainerManifestAccept(accept) legacyCacheKey := h.containerManifestCacheKey(upstream.URL, "library/nginx", "latest", accept) - cacheKey := h.containerManifestCacheKey(upstream.URL, "library/nginx", "latest", cacheAccept) + cacheKey := h.containerManifestSharedCacheKey(upstream.URL, "library/nginx", "latest", cacheAccept) if legacyCacheKey == cacheKey { - t.Fatal("legacy and normalized cache keys are equal") + t.Fatal("legacy and shared cache keys are equal") } request := httptest.NewRequest(http.MethodGet, "/library/nginx/manifests/latest", nil) @@ -1108,7 +1108,7 @@ func TestContainerHandler_ManifestMigratesLegacyAcceptCacheKey(t *testing.T) { t.Fatalf("load migrated manifest: %v", err) } if migrated == nil { - t.Error("normalized cache entry was not created") + t.Error("shared cache entry was not created") } } From cc18c889ad1b1e25e3bec4fc71b446668ffc2ed0 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Mon, 12 Oct 2026 00:02:31 +0200 Subject: [PATCH 2/3] proxy mirror now builds its upstream clients the same way as serve, so it follows OCI Bearer challenges, uses configured upstream credentials and the SSRF guard, and artifact downloads have no overall time limit. Before this the CLI used a plain client and Docker Hub answered every request with 401. --- cmd/proxy/main.go | 8 +--- internal/server/mirror_proxy_test.go | 64 ++++++++++++++++++++++++++++ internal/server/server.go | 48 ++++++++++++++------- 3 files changed, 99 insertions(+), 21 deletions(-) create mode 100644 internal/server/mirror_proxy_test.go diff --git a/cmd/proxy/main.go b/cmd/proxy/main.go index 5b625631..acd32cc5 100644 --- a/cmd/proxy/main.go +++ b/cmd/proxy/main.go @@ -113,11 +113,9 @@ import ( "github.com/git-pkgs/proxy/internal/config" "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/denylist" - "github.com/git-pkgs/proxy/internal/handler" "github.com/git-pkgs/proxy/internal/mirror" "github.com/git-pkgs/proxy/internal/server" "github.com/git-pkgs/proxy/internal/storage" - "github.com/git-pkgs/registries/fetch" ) const defaultTopN = 10 @@ -518,10 +516,8 @@ func runMirror() { os.Exit(1) //nolint:gocritic // db closed above } - // Build proxy (reuses same pipeline as serve) - fetcher := fetch.NewFetcher() - resolver := fetch.NewResolver() - proxy := handler.NewProxy(db, store, fetcher, resolver, logger) + // Build proxy with the same upstream clients and authentication as serve + proxy := server.NewMirrorProxy(cfg, db, store, logger) proxy.Denylist, err = denylist.New(cfg.Denylist.Packages) if err != nil { _ = db.Close() diff --git a/internal/server/mirror_proxy_test.go b/internal/server/mirror_proxy_test.go new file mode 100644 index 00000000..2df395af --- /dev/null +++ b/internal/server/mirror_proxy_test.go @@ -0,0 +1,64 @@ +package server + +import ( + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + + "github.com/git-pkgs/proxy/internal/config" +) + +// newTokenRegistry answers 401 with a Bearer challenge until the request +// carries the token its realm hands out. +func newTokenRegistry(t *testing.T) *httptest.Server { + t.Helper() + var registry *httptest.Server + registry = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/token": + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"token":"pull-token"}`) + case "/v2/library/app/manifests/1.0": + if r.Header.Get("Authorization") != "Bearer pull-token" { + w.Header().Set("WWW-Authenticate", `Bearer realm="`+registry.URL+`/token",service="registry",scope="repository:library/app:pull"`) + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + w.Header().Set("Content-Type", "application/vnd.oci.image.manifest.v1+json") + _, _ = io.WriteString(w, `{"schemaVersion":2}`) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(registry.Close) + return registry +} + +func TestNewMirrorProxyFollowsBearerChallenge(t *testing.T) { + registry := newTokenRegistry(t) + cfg := config.Default() + cfg.Upstream.AllowLoopback = true + proxy := NewMirrorProxy(cfg, nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) + + resp, err := proxy.HTTPClient.Get(registry.URL + "/v2/library/app/manifests/1.0") + if err != nil { + t.Fatalf("GET manifest: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + t.Errorf("status = %d, want 200 after the Bearer challenge", resp.StatusCode) + } +} + +func TestNewMirrorProxyRefusesLoopbackByDefault(t *testing.T) { + registry := newTokenRegistry(t) + proxy := NewMirrorProxy(config.Default(), nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) + + resp, err := proxy.HTTPClient.Get(registry.URL + "/v2/library/app/manifests/1.0") + if err == nil { + _ = resp.Body.Close() + t.Fatal("GET to a loopback registry succeeded without upstream.allow_loopback") + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 380892a7..de6204c1 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -220,22 +220,10 @@ func (s *Server) Start(listeners ...net.Listener) error { } func (s *Server) serve(listener net.Listener) error { - // Use one authentication-aware transport for metadata and artifacts so - // configured credentials and cached OCI challenges apply consistently. - safeClient := newUpstreamClient(s.cfg.Upstream) - baseTransport := safeClient.Transport - if s.accessLog != nil { - baseTransport = upstreamhttp.NewAccessLogTransport(baseTransport, s.accessLog, s.logger) - } - authTransport := upstreamhttp.NewTransport(baseTransport, upstreamhttp.AuthFunc(s.authForURL)) - metadataClient := *safeClient - metadataClient.Timeout = s.cfg.ParseHTTPTimeout() - metadataClient.Transport = authTransport - artifactClient := metadataClient - artifactClient.Timeout = serverWriteTimeout + metadataClient, artifactClient := s.newUpstreamClients(serverWriteTimeout) // Create shared components with circuit breaker. - baseFetcher := fetch.NewFetcher(fetch.WithHTTPClient(&artifactClient)) + baseFetcher := fetch.NewFetcher(fetch.WithHTTPClient(artifactClient)) fetcher := fetch.NewCircuitBreakerFetcher(baseFetcher) s.breakers = newBreakerMonitor(fetcher, s.logger) resolver := fetch.NewResolver() @@ -245,7 +233,7 @@ func (s *Server) serve(listener net.Listener) error { Packages: s.cfg.Cooldown.NormalizedPackages(), } proxy := handler.NewProxy(s.db, s.storage, fetcher, resolver, s.logger) - proxy.HTTPClient = &metadataClient + proxy.HTTPClient = metadataClient proxy.AuthForURL = s.authForURL cooldownPolicy, err := cooldownpolicy.New(cd, s.cfg.Cooldown.PackagePatterns) if err != nil { @@ -481,6 +469,36 @@ func (s *Server) mountProtocolHandlers(r chi.Router, proxy *handler.Proxy) { // configureScanning builds the scanner group from cfg and wires it into // proxy, returning the group so the caller can decide whether to mount the // internal scan-fetch route. +// newUpstreamClients builds the metadata and artifact clients. Both share one +// authentication-aware transport so configured credentials and cached OCI +// challenges apply consistently. An artifactTimeout of zero means no limit. +func (s *Server) newUpstreamClients(artifactTimeout time.Duration) (metadataClient, artifactClient *http.Client) { + safeClient := newUpstreamClient(s.cfg.Upstream) + baseTransport := safeClient.Transport + if s.accessLog != nil { + baseTransport = upstreamhttp.NewAccessLogTransport(baseTransport, s.accessLog, s.logger) + } + authTransport := upstreamhttp.NewTransport(baseTransport, upstreamhttp.AuthFunc(s.authForURL)) + metadata := *safeClient + metadata.Timeout = s.cfg.ParseHTTPTimeout() + metadata.Transport = authTransport + artifact := metadata + artifact.Timeout = artifactTimeout + return &metadata, &artifact +} + +// NewMirrorProxy builds the proxy for the mirror command with the same +// upstream clients, SSRF guard and upstream authentication as serve. Artifact +// downloads have no overall time limit; the caller cancels through ctx. +func NewMirrorProxy(cfg *config.Config, db *database.DB, store storage.Storage, logger *slog.Logger) *handler.Proxy { + s := &Server{cfg: cfg, db: db, storage: store, logger: logger, ecr: newECRTokens(logger)} + metadataClient, artifactClient := s.newUpstreamClients(0) + proxy := handler.NewProxy(db, store, fetch.NewFetcher(fetch.WithHTTPClient(artifactClient)), fetch.NewResolver(), logger) + proxy.HTTPClient = metadataClient + proxy.AuthForURL = s.authForURL + return proxy +} + 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 { From 7b6c64af0a07ef062a0472e03d215cbc1bd2e619 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Mon, 12 Oct 2026 00:20:10 +0200 Subject: [PATCH 3/3] proxy mirror and the mirror API now pre-fetch container images from pkg:docker and pkg:oci PURLs: the mirror walks index, platform manifests (with their attestations) and blobs through the container handler, so a later pull is served from cache even when the upstream is down. A --platform filter (platforms in the API) limits multi-arch images, repository_url only resolves to configured registries, and names, tags and digests are validated before anything is fetched. --- README.md | 3 + cmd/proxy/main.go | 24 +- docs/architecture.md | 2 +- docs/configuration.md | 27 ++ internal/handler/container_mirror.go | 373 ++++++++++++++++++++++ internal/handler/container_mirror_test.go | 295 +++++++++++++++++ internal/mirror/image.go | 126 ++++++++ internal/mirror/image_test.go | 174 ++++++++++ internal/mirror/job.go | 10 +- internal/mirror/mirror.go | 8 + internal/mirror/source.go | 22 +- internal/server/mirror_proxy_test.go | 4 +- internal/server/server.go | 31 +- 13 files changed, 1076 insertions(+), 23 deletions(-) create mode 100644 internal/handler/container_mirror.go create mode 100644 internal/handler/container_mirror_test.go create mode 100644 internal/mirror/image.go create mode 100644 internal/mirror/image_test.go diff --git a/README.md b/README.md index 06cd0edb..05cc362c 100644 --- a/README.md +++ b/README.md @@ -890,6 +890,9 @@ proxy mirror --dry-run pkg:npm/lodash # Control parallelism proxy mirror --concurrency 8 pkg:npm/lodash@4.17.21 + +# Mirror a container image for one platform +proxy mirror --platform linux/amd64 pkg:docker/library/alpine@3.20 ``` The mirror command accepts the same storage and database flags as `serve`. Already-cached artifacts are skipped. diff --git a/cmd/proxy/main.go b/cmd/proxy/main.go index acd32cc5..b4208d15 100644 --- a/cmd/proxy/main.go +++ b/cmd/proxy/main.go @@ -419,6 +419,7 @@ func runMirror() { sbomPath := fs.String("sbom", "", "Path to CycloneDX or SPDX SBOM file") concurrency := fs.Int("concurrency", 4, "Number of parallel downloads") //nolint:mnd // default concurrency dryRun := fs.Bool("dry-run", false, "Show what would be mirrored without downloading") + platforms := fs.String("platform", "", "Comma-separated os/arch[/variant] list limiting which platforms of container images are mirrored") fs.Usage = func() { fmt.Fprintf(os.Stderr, "git-pkgs proxy - Pre-populate cache\n\n") @@ -426,13 +427,22 @@ func runMirror() { fmt.Fprintf(os.Stderr, "Examples:\n") fmt.Fprintf(os.Stderr, " proxy mirror pkg:npm/lodash@4.17.21\n") fmt.Fprintf(os.Stderr, " proxy mirror --sbom sbom.cdx.json\n") - fmt.Fprintf(os.Stderr, " proxy mirror pkg:npm/lodash # all versions\n\n") + fmt.Fprintf(os.Stderr, " proxy mirror pkg:npm/lodash # all versions\n") + fmt.Fprintf(os.Stderr, " proxy mirror --platform linux/amd64 pkg:docker/library/alpine@3.20\n\n") fmt.Fprintf(os.Stderr, "Flags:\n") fs.PrintDefaults() } _ = fs.Parse(os.Args[1:]) purls := fs.Args() + var platformList []string + if *platforms != "" { + platformList = []string{*platforms} + } + if _, err := mirror.ParsePlatforms(platformList); err != nil { + fmt.Fprintf(os.Stderr, "error: %v\n", err) + os.Exit(1) + } // Determine source var source mirror.Source @@ -443,9 +453,9 @@ func runMirror() { fmt.Fprintf(os.Stderr, "error reading SBOM %s: %v\n", *sbomPath, err) os.Exit(1) } - source = &mirror.SBOMSource{Data: data, Name: *sbomPath} + source = &mirror.SBOMSource{Data: data, Name: *sbomPath, Platforms: platformList} case len(purls) > 0: - source = &mirror.PURLSource{PURLs: purls} + source = &mirror.PURLSource{PURLs: purls, Platforms: platformList} default: fmt.Fprintf(os.Stderr, "error: provide PURLs or --sbom\n") fs.Usage() @@ -517,7 +527,7 @@ func runMirror() { } // Build proxy with the same upstream clients and authentication as serve - proxy := server.NewMirrorProxy(cfg, db, store, logger) + proxy, images := server.NewMirrorProxy(cfg, db, store, logger) proxy.Denylist, err = denylist.New(cfg.Denylist.Packages) if err != nil { _ = db.Close() @@ -529,6 +539,12 @@ func runMirror() { proxy.MetadataMaxSize = cfg.ParseMetadataMaxSize() m := mirror.New(proxy, db, store, logger, *concurrency) + m.Images = images + if cfg.Scanning.Enabled { + // Scanners fetch staged layers from the running server, which this + // command does not start. + m.Images = mirror.UnavailableImages{Reason: "container images cannot be mirrored by this command while scanning is enabled; use the mirror API of the running server"} + } ctx, cancel := context.WithCancel(context.Background()) go func() { diff --git a/docs/architecture.md b/docs/architecture.md index 0d37a60a..7529d5e0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -316,7 +316,7 @@ Package metadata enrichment. Fetches license, description, homepage, repository ### `internal/mirror` -Selective package mirroring for pre-populating the proxy cache. Supports multiple input sources: individual PURLs (versioned or unversioned), CycloneDX/SPDX SBOM files, and full registry enumeration. Uses a bounded worker pool backed by `errgroup` to download artifacts in parallel, reusing `handler.Proxy.GetOrFetchArtifact()` for the actual fetch-and-cache work. +Selective package mirroring for pre-populating the proxy cache. Supports multiple input sources: individual PURLs (versioned or unversioned), CycloneDX/SPDX SBOM files, and full registry enumeration. Uses a bounded worker pool backed by `errgroup` to download artifacts in parallel, reusing `handler.Proxy.GetOrFetchArtifact()` for the actual fetch-and-cache work. Container images (`pkg:docker`, `pkg:oci`) go through `handler.ContainerHandler.MirrorImage()` instead, which walks index, platform manifests and blobs and stores them under the keys a pull reads. The package also provides a `MetadataCache` for storing raw upstream metadata blobs so the proxy can serve metadata responses offline. The `JobStore` manages async mirror jobs exposed via the `/api/mirror` endpoints. diff --git a/docs/configuration.md b/docs/configuration.md index 675d45af..4986a7db 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -829,6 +829,14 @@ curl -X POST http://localhost:8080/api/mirror \ -d '{"sbom":{"bomFormat":"CycloneDX","components":[{"purl":"pkg:npm/lodash@4.17.21"}]}}' ``` +`platforms` limits which platforms of container images are mirrored, like the command's `--platform` flag: + +```bash +curl -X POST http://localhost:8080/api/mirror \ + -H "Content-Type: application/json" \ + -d '{"purls":["pkg:docker/library/alpine@3.20"],"platforms":["linux/amd64"]}' +``` + ## Mirror Command The `proxy mirror` command pre-populates the cache from various sources. It accepts the same storage and database flags as `serve`. @@ -838,6 +846,7 @@ The `proxy mirror` command pre-populates the cache from various sources. It acce | `--sbom` | | Path to CycloneDX or SPDX SBOM file | | `--concurrency` | `4` | Number of parallel downloads | | `--dry-run` | `false` | Show what would be mirrored without downloading | +| `--platform` | | Comma-separated `os/arch[/variant]` list limiting which platforms of container images are mirrored | | `--config` | | Path to configuration file | | `--storage-url` | | Storage URL | | `--database-driver` | | Database driver | @@ -850,6 +859,24 @@ Positional arguments are treated as PURLs: proxy mirror pkg:npm/lodash@4.17.21 pkg:cargo/serde@1.0.0 ``` +### Container images + +`pkg:docker` and `pkg:oci` PURLs mirror a container image the way a pull does: the manifest or index, the platform manifests, then config and layer blobs. Everything is stored under the keys a pull through `/v2` reads, so a later `docker pull` or containerd pull is served from cache, also when the upstream is unreachable. + +```bash +proxy mirror --platform linux/amd64,linux/arm64 pkg:docker/library/alpine@3.20 +proxy mirror 'pkg:docker/containerd/alpine@3.14.0?repository_url=ghcr.io' +proxy mirror 'pkg:oci/debian@sha256:...?repository_url=docker.io/library/debian&tag=bookworm' +``` + +- The version is a tag or a digest. Unversioned image PURLs fail; mirroring all tags is not supported. +- Without `repository_url` the name is routed like a pull without containerd's `ns` parameter: the default OCI registry, repository prefix routes and `upstream/{name}/`. Single-segment Docker Hub names get the `library/` prefix. +- `repository_url` is resolved like containerd's `ns` parameter, so it has to name the default registry, Docker Hub, a configured `upstream.oci` host or a host listed in `upstream.oci_mirrors`. Other hosts are refused. +- `pkg:oci` needs `repository_url`. With `tag=` the tag is mirrored as well; if it now points to another digest, both are mirrored and a warning is logged. `arch=` limits the platforms when `--platform` is not given. +- Without `--platform` every platform of an index is mirrored, including Windows images, which can be large. Attestation manifests of the selected platforms are mirrored too, because Docker fetches them on every pull. +- Non-distributable (foreign) layers are skipped. +- While `scanning.enabled` is set, the command refuses images, because scanners fetch staged layers from the running server. Use the mirror API instead. + ## Docker ### SQLite with Local Storage diff --git a/internal/handler/container_mirror.go b/internal/handler/container_mirror.go new file mode 100644 index 00000000..f9e3d495 --- /dev/null +++ b/internal/handler/container_mirror.go @@ -0,0 +1,373 @@ +package handler + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/url" + "regexp" + "slices" + "strings" +) + +const ( + // maxMirrorIndexDepth bounds how deep nested indexes are followed. + maxMirrorIndexDepth = 2 + // mirrorErrorBodyLimit bounds how much of an error response is quoted. + mirrorErrorBodyLimit = 200 + + dockerReferenceTypeAnnotation = "vnd.docker.reference.type" + dockerReferenceDigestAnnotation = "vnd.docker.reference.digest" + attestationManifestReference = "attestation-manifest" +) + +var ( + // Grammar from the OCI distribution spec. + mirrorNamePattern = regexp.MustCompile(`^[a-z0-9]+((\.|_|__|-+)[a-z0-9]+)*(/[a-z0-9]+((\.|_|__|-+)[a-z0-9]+)*)*$`) + mirrorTagPattern = regexp.MustCompile(`^[a-zA-Z0-9_][a-zA-Z0-9._-]{0,127}$`) + mirrorDigestPattern = regexp.MustCompile(`^sha256:[a-f0-9]{64}$`) +) + +// dockerHubMirrorAliases are registry hosts that name Docker Hub in image +// references. hub.docker.com is the default registry of the docker PURL type. +var dockerHubMirrorAliases = []string{"docker.io", "index.docker.io", "registry-1.docker.io", "hub.docker.com"} //nolint:gochecknoglobals // fixed alias list + +// ImageRef names an image to mirror. +type ImageRef struct { + // Registry is the registry host the image reference names, as containerd + // sends it in the ns query parameter. Empty routes Name like a pull + // without ns. + Registry string + // Name is the repository name. + Name string + // References are the tags and digests to mirror. + References []string +} + +// Platform selects the entries of an image index to mirror. Empty fields +// match any value. +type Platform struct { + OS string + Architecture string + Variant string +} + +// ParsePlatform parses os/architecture[/variant]. +func ParsePlatform(value string) (Platform, error) { + os, rest, _ := strings.Cut(value, "/") + architecture, variant, _ := strings.Cut(rest, "/") + if os == "" || architecture == "" || strings.Contains(variant, "/") { + return Platform{}, fmt.Errorf("invalid platform %q, want os/architecture[/variant]", value) + } + return Platform{OS: os, Architecture: architecture, Variant: variant}, nil +} + +func (p Platform) matches(entry *ociPlatform) bool { + if entry == nil { + return false + } + if p.OS != "" && p.OS != entry.OS { + return false + } + if p.Architecture != "" && p.Architecture != entry.Architecture { + return false + } + return p.Variant == "" || p.Variant == platformVariant(entry) +} + +// platformVariant fills in the variant arm64 images usually leave out. +func platformVariant(entry *ociPlatform) string { + if entry.Variant == "" && entry.Architecture == "arm64" { + return "v8" + } + return entry.Variant +} + +// ImageResult reports what mirroring one image did. +type ImageResult struct { + // Fetched counts manifests and blobs that were not cached before. + Fetched int + // Bytes is the size of the blobs fetched for this image. + Bytes int64 + Warnings []string +} + +type ociDescriptor struct { + MediaType string `json:"mediaType"` + Digest string `json:"digest"` + URLs []string `json:"urls"` + Platform *ociPlatform `json:"platform"` + Annotations map[string]string `json:"annotations"` +} + +type ociPlatform struct { + OS string `json:"os"` + Architecture string `json:"architecture"` + Variant string `json:"variant"` +} + +type ociManifest struct { + SchemaVersion int `json:"schemaVersion"` + MediaType string `json:"mediaType"` + Manifests []ociDescriptor `json:"manifests"` + Config *ociDescriptor `json:"config"` + Layers []ociDescriptor `json:"layers"` +} + +type imageMirror struct { + h *ContainerHandler + ref ImageRef + registryURL string + upstreamName string + cacheName string + platforms []Platform + seen map[string]bool + result ImageResult +} + +// MirrorImage fetches an image the way a client pull does: the manifest or +// index for each reference, the platform manifests, then config and layer +// blobs. Everything is stored under the keys a later pull through this handler +// reads, so the pull is served from cache. Platforms limit which index entries +// are followed; without them every platform is mirrored. +func (h *ContainerHandler) MirrorImage(ctx context.Context, ref ImageRef, platforms []Platform) (ImageResult, error) { + ref, err := normalizeImageRef(ref, h.registryURL) + if err != nil { + return ImageResult{}, err + } + route := &http.Request{URL: &url.URL{RawQuery: url.Values{namespaceQueryParam: {ref.Registry}}.Encode()}} + if ref.Registry == "" { + route.URL.RawQuery = "" + } + registryURL, upstreamName, cacheName, ok := h.registryForRequest(route, ref.Name) + if !ok { + if ref.Registry != "" { + return ImageResult{}, fmt.Errorf("registry %q is not configured as an OCI upstream", ref.Registry) + } + return ImageResult{}, fmt.Errorf("no OCI upstream for %q", ref.Name) + } + m := &imageMirror{ + h: h, + ref: ref, + registryURL: registryURL, + upstreamName: upstreamName, + cacheName: cacheName, + platforms: platforms, + seen: make(map[string]bool), + } + var digests []string + for _, reference := range ref.References { + digest, err := m.mirrorManifest(ctx, reference, 0) + if err != nil { + return m.result, fmt.Errorf("%s:%s: %w", ref.Name, reference, err) + } + digests = append(digests, digest) + } + if len(digests) == 2 && digests[0] != digests[1] && mirrorDigestPattern.MatchString(ref.References[0]) { + m.result.Warnings = append(m.result.Warnings, fmt.Sprintf("tag %s now points to %s, not %s; both were mirrored", + ref.References[1], digests[1], ref.References[0])) + } + return m.result, nil +} + +// normalizeImageRef validates the reference before any network access and +// adds the library/ prefix Docker Hub clients use for official images. +func normalizeImageRef(ref ImageRef, defaultRegistryURL string) (ImageRef, error) { + if slices.Contains(dockerHubMirrorAliases, strings.ToLower(ref.Registry)) { + ref.Registry = "docker.io" + } + dockerHub := ref.Registry == "docker.io" || (ref.Registry == "" && defaultRegistryURL == dockerHubRegistry) + if dockerHub && !strings.Contains(ref.Name, "/") { + ref.Name = "library/" + ref.Name + } + if !mirrorNamePattern.MatchString(ref.Name) { + return ref, fmt.Errorf("invalid repository name %q", ref.Name) + } + if len(ref.References) == 0 { + return ref, errors.New("a tag or digest is required; mirroring all tags is not supported") + } + for _, reference := range ref.References { + if !mirrorTagPattern.MatchString(reference) && !mirrorDigestPattern.MatchString(reference) { + return ref, fmt.Errorf("invalid tag or digest %q", reference) + } + } + return ref, nil +} + +// mirrorManifest stores the manifest for reference through the handler's own +// manifest route, then follows what it references. It returns the manifest +// digest. +func (m *imageMirror) mirrorManifest(ctx context.Context, reference string, depth int) (string, error) { + cacheAccept := normalizeContainerManifestAccept(containerManifestAccept(&http.Request{Header: http.Header{}})) + before := m.h.loadContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, reference, cacheAccept, cacheAccept) + + target := "/" + m.ref.Name + "/manifests/" + reference + if m.ref.Registry != "" { + target += "?" + url.Values{namespaceQueryParam: {m.ref.Registry}}.Encode() + } + request, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) + if err != nil { + return "", err + } + response := newMirrorResponse() + m.h.Routes().ServeHTTP(response, request) + if response.status != http.StatusOK { + return "", fmt.Errorf("manifest %s: status %d: %s", reference, response.status, response.errorText()) + } + if response.header.Get("Warning") != "" { + m.result.Warnings = append(m.result.Warnings, fmt.Sprintf("manifest %s served from stale cache, upstream unreachable", reference)) + } + digest := response.header.Get("Docker-Content-Digest") + if !mirrorDigestPattern.MatchString(digest) { + digest = sha256Digest(response.body.Bytes()) + } + if before == nil || before.contentDigest != digest { + m.result.Fetched++ + } + if m.seen[digest] { + return digest, nil + } + m.seen[digest] = true + + var manifest ociManifest + if err := json.Unmarshal(response.body.Bytes(), &manifest); err != nil { + return "", fmt.Errorf("manifest %s: %w", reference, err) + } + if manifest.SchemaVersion == 1 { + return "", fmt.Errorf("manifest %s: schema 1 manifests are not supported", reference) + } + if manifest.Manifests != nil { + return digest, m.mirrorIndex(ctx, reference, &manifest, depth) + } + return digest, m.mirrorBlobs(ctx, &manifest) +} + +// mirrorIndex follows the index entries for the wanted platforms. Docker +// fetches the attestation manifests of an index on every pull, so those of the +// selected platforms are mirrored too. +func (m *imageMirror) mirrorIndex(ctx context.Context, reference string, index *ociManifest, depth int) error { + if depth >= maxMirrorIndexDepth { + return fmt.Errorf("index %s: nested deeper than %d levels", reference, maxMirrorIndexDepth) + } + var selected, attestations []ociDescriptor + selectedDigests := make(map[string]bool) + for _, entry := range index.Manifests { + if !mirrorDigestPattern.MatchString(entry.Digest) { + return fmt.Errorf("index %s: invalid digest %q", reference, entry.Digest) + } + if entry.Annotations[dockerReferenceTypeAnnotation] == attestationManifestReference { + attestations = append(attestations, entry) + continue + } + if m.wantsPlatform(entry.Platform) { + selected = append(selected, entry) + selectedDigests[entry.Digest] = true + } + } + if len(selected) == 0 && len(m.platforms) > 0 { + return fmt.Errorf("index %s: no manifest for the requested platforms", reference) + } + for _, entry := range attestations { + if len(m.platforms) == 0 || selectedDigests[entry.Annotations[dockerReferenceDigestAnnotation]] { + selected = append(selected, entry) + } + } + for _, entry := range selected { + if _, err := m.mirrorManifest(ctx, entry.Digest, depth+1); err != nil { + return err + } + } + return nil +} + +func (m *imageMirror) wantsPlatform(platform *ociPlatform) bool { + if len(m.platforms) == 0 || platform == nil { + // Entries without a platform are nested indexes or artifacts; a + // filter cannot judge them, so they are followed. + return true + } + for _, want := range m.platforms { + if want.matches(platform) { + return true + } + } + return false +} + +func (m *imageMirror) mirrorBlobs(ctx context.Context, manifest *ociManifest) error { + blobs := manifest.Layers + if manifest.Config != nil { + blobs = append([]ociDescriptor{*manifest.Config}, blobs...) + } + for _, blob := range blobs { + if len(blob.URLs) > 0 || strings.Contains(blob.MediaType, "nondistributable") || strings.Contains(blob.MediaType, "foreign") { + m.result.Warnings = append(m.result.Warnings, fmt.Sprintf("skipped non-distributable layer %s", blob.Digest)) + continue + } + if !mirrorDigestPattern.MatchString(blob.Digest) { + return fmt.Errorf("invalid blob digest %q", blob.Digest) + } + if m.seen[blob.Digest] { + continue + } + m.seen[blob.Digest] = true + if err := m.mirrorBlob(ctx, blob.Digest); err != nil { + return err + } + } + return nil +} + +// mirrorBlob stores a blob under the same identity handleBlobDownload uses. +func (m *imageMirror) mirrorBlob(ctx context.Context, digest string) error { + proxy := m.h.proxy + cached, err := proxy.GetCachedArtifact(ctx, "oci", m.cacheName, digest, digest) + if err != nil { + return fmt.Errorf("blob %s: %w", digest, err) + } + if cached != nil { + if cached.Reader != nil { + _ = cached.Reader.Close() + } + return nil + } + result, err := proxy.GetOrFetchArtifactFromURLWithDigest(ctx, "oci", m.cacheName, digest, digest, + fmt.Sprintf("%s/v2/%s/blobs/%s", m.registryURL, m.upstreamName, digest), digest) + if err != nil { + return fmt.Errorf("blob %s: %w", digest, err) + } + if result.Reader != nil { + _ = result.Reader.Close() + } + m.result.Fetched++ + m.result.Bytes += result.Artifact.Size + return nil +} + +// mirrorResponse collects the handler's manifest response in memory. +type mirrorResponse struct { + header http.Header + status int + body bytes.Buffer +} + +func newMirrorResponse() *mirrorResponse { + return &mirrorResponse{header: make(http.Header), status: http.StatusOK} +} + +func (r *mirrorResponse) Header() http.Header { return r.header } + +func (r *mirrorResponse) WriteHeader(status int) { r.status = status } + +func (r *mirrorResponse) Write(p []byte) (int, error) { return r.body.Write(p) } + +func (r *mirrorResponse) errorText() string { + text := strings.TrimSpace(r.body.String()) + if len(text) > mirrorErrorBodyLimit { + text = text[:mirrorErrorBodyLimit] + "..." + } + return text +} diff --git a/internal/handler/container_mirror_test.go b/internal/handler/container_mirror_test.go new file mode 100644 index 00000000..fc4e8a60 --- /dev/null +++ b/internal/handler/container_mirror_test.go @@ -0,0 +1,295 @@ +package handler + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync/atomic" + "testing" + + "github.com/git-pkgs/registries/fetch" +) + +// fakeImageRegistry serves manifests and blobs of a small multi-arch image, +// a single-arch image and a moved tag. It fails with 503 once offline. +type fakeImageRegistry struct { + server *httptest.Server + offline atomic.Bool + manifests map[string]fakeManifest + blobs map[string]string + blobGets atomic.Int32 +} + +type fakeImage struct { + indexDigest string + amd64Digest string + arm64Digest string + amd64Layer string + arm64Layer string + singleDigest string + singleLayer string + // attestationLayers holds the amd64 and arm64 provenance statements. + attestationLayers []string +} + +func newFakeImageRegistry(t *testing.T) (*fakeImageRegistry, fakeImage) { + t.Helper() + const repository = "team/app" + r := &fakeImageRegistry{manifests: make(map[string]fakeManifest), blobs: make(map[string]string)} + var image fakeImage + + addBlob := func(content string) string { + digest := sha256Digest([]byte(content)) + r.blobs["/v2/"+repository+"/blobs/"+digest] = content + return digest + } + addManifest := func(contentType string, doc any, tags ...string) string { + body, err := json.Marshal(doc) + if err != nil { + t.Fatal(err) + } + digest := sha256Digest(body) + for _, reference := range append(tags, digest) { + r.manifests["/v2/"+repository+"/manifests/"+reference] = fakeManifest{contentType: contentType, body: string(body)} + } + return digest + } + imageManifest := func(platform string) (string, string) { + config := addBlob(`{"architecture":"` + platform + `"}`) + layer := addBlob("layer for " + platform) + digest := addManifest(ociManifestMediaType, map[string]any{ + "schemaVersion": 2, "mediaType": ociManifestMediaType, + "config": map[string]any{"mediaType": "application/vnd.oci.image.config.v1+json", "digest": config}, + "layers": []map[string]any{ + {"mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", "digest": layer}, + {"mediaType": "application/vnd.oci.image.layer.nondistributable.v1.tar+gzip", "digest": sha256Digest([]byte("foreign")), "urls": []string{"https://example.invalid/layer"}}, + }, + }) + return digest, layer + } + + image.amd64Digest, image.amd64Layer = imageManifest("amd64") + image.arm64Digest, image.arm64Layer = imageManifest("arm64") + attestation := func(platformDigest, platform string) map[string]any { + statement := addBlob("provenance for " + platform) + digest := addManifest(ociManifestMediaType, map[string]any{ + "schemaVersion": 2, "mediaType": ociManifestMediaType, + "config": map[string]any{"mediaType": "application/vnd.oci.image.config.v1+json", "digest": addBlob("{}")}, + "layers": []map[string]any{{"mediaType": "application/vnd.in-toto+json", "digest": statement}}, + }) + image.attestationLayers = append(image.attestationLayers, statement) + return map[string]any{ + "mediaType": ociManifestMediaType, "digest": digest, + "platform": map[string]string{"os": "unknown", "architecture": "unknown"}, + "annotations": map[string]string{ + dockerReferenceTypeAnnotation: attestationManifestReference, + dockerReferenceDigestAnnotation: platformDigest, + }, + } + } + image.indexDigest = addManifest(ociImageIndexMediaType, map[string]any{ + "schemaVersion": 2, "mediaType": ociImageIndexMediaType, + "manifests": []map[string]any{ + {"mediaType": ociManifestMediaType, "digest": image.amd64Digest, "platform": map[string]string{"os": "linux", "architecture": "amd64"}}, + {"mediaType": ociManifestMediaType, "digest": image.arm64Digest, "platform": map[string]string{"os": "linux", "architecture": "arm64", "variant": "v8"}}, + attestation(image.amd64Digest, "amd64"), + attestation(image.arm64Digest, "arm64"), + }, + }, "multi") + + image.singleDigest, image.singleLayer = imageManifest("single") + r.manifests["/v2/"+repository+"/manifests/single"] = r.manifests["/v2/"+repository+"/manifests/"+image.singleDigest] + + r.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if r.offline.Load() { + http.Error(w, "offline", http.StatusServiceUnavailable) + return + } + if doc, ok := r.manifests[req.URL.Path]; ok { + w.Header().Set("Content-Type", doc.contentType) + w.Header().Set("Docker-Content-Digest", sha256Digest([]byte(doc.body))) + _, _ = io.WriteString(w, doc.body) + return + } + if blob, ok := r.blobs[req.URL.Path]; ok { + r.blobGets.Add(1) + _, _ = io.WriteString(w, blob) + return + } + http.NotFound(w, req) + })) + t.Cleanup(r.server.Close) + return r, image +} + +func newMirrorTestProxy(t *testing.T, registry *fakeImageRegistry) *Proxy { + t.Helper() + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = registry.server.Client() + proxy.Fetcher = fetch.NewFetcher(fetch.WithHTTPClient(registry.server.Client()), fetch.WithMaxRetries(0)) + proxy.MetadataTTL = 0 + return proxy +} + +func pull(t *testing.T, h *ContainerHandler, target, accept string) *httptest.ResponseRecorder { + t.Helper() + request := httptest.NewRequest(http.MethodGet, target, nil) + if accept != "" { + request.Header.Set("Accept", accept) + } + response := httptest.NewRecorder() + h.Routes().ServeHTTP(response, request) + return response +} + +func TestMirrorImageThenPullOffline(t *testing.T) { + registry, image := newFakeImageRegistry(t) + proxy := newMirrorTestProxy(t, registry) + mirrorHandler := NewContainerHandlerWithRegistry(proxy, "http://proxy.example", registry.server.URL) + + result, err := mirrorHandler.MirrorImage(context.Background(), ImageRef{Name: "team/app", References: []string{"multi", "single"}}, nil) + if err != nil { + t.Fatalf("MirrorImage: %v", err) + } + if len(result.Warnings) == 0 || !strings.Contains(strings.Join(result.Warnings, "\n"), "non-distributable") { + t.Errorf("warnings = %v, want the skipped foreign layer", result.Warnings) + } + registry.offline.Store(true) + + // A fresh handler instance, as after a restart, serves the pull. + h := NewContainerHandlerWithRegistry(proxy, "http://proxy.example", registry.server.URL) + for _, accept := range []string{dockerManifestAccept, containerdManifestAccept} { + for _, reference := range []string{"multi", image.indexDigest, image.amd64Digest, image.arm64Digest, "single", image.singleDigest} { + if response := pull(t, h, "/team/app/manifests/"+reference, accept); response.Code != http.StatusOK { + t.Errorf("offline manifest %s with %.30q: status %d: %s", reference, accept, response.Code, response.Body.String()) + } + } + } + for _, layer := range append([]string{image.amd64Layer, image.arm64Layer, image.singleLayer}, image.attestationLayers...) { + if response := pull(t, h, "/team/app/blobs/"+layer, ""); response.Code != http.StatusOK { + t.Errorf("offline blob %s: status %d", layer, response.Code) + } + } + + registry.offline.Store(false) + gets := registry.blobGets.Load() + again, err := mirrorHandler.MirrorImage(context.Background(), ImageRef{Name: "team/app", References: []string{"multi"}}, nil) + if err != nil { + t.Fatalf("second MirrorImage: %v", err) + } + if again.Fetched != 0 || registry.blobGets.Load() != gets { + t.Errorf("second run fetched %d items and %d blobs, want nothing new", again.Fetched, registry.blobGets.Load()-gets) + } +} + +func TestMirrorImagePlatformFilter(t *testing.T) { + registry, image := newFakeImageRegistry(t) + proxy := newMirrorTestProxy(t, registry) + h := NewContainerHandlerWithRegistry(proxy, "http://proxy.example", registry.server.URL) + + platform, err := ParsePlatform("linux/arm64") + if err != nil { + t.Fatal(err) + } + if _, err := h.MirrorImage(context.Background(), ImageRef{Name: "team/app", References: []string{"multi"}}, []Platform{platform}); err != nil { + t.Fatalf("MirrorImage: %v", err) + } + registry.offline.Store(true) + if response := pull(t, h, "/team/app/blobs/"+image.arm64Layer, ""); response.Code != http.StatusOK { + t.Errorf("arm64 layer status = %d, want 200", response.Code) + } + if response := pull(t, h, "/team/app/blobs/"+image.attestationLayers[1], ""); response.Code != http.StatusOK { + t.Errorf("arm64 attestation status = %d, want 200", response.Code) + } + for _, layer := range []string{image.amd64Layer, image.attestationLayers[0]} { + if response := pull(t, h, "/team/app/blobs/"+layer, ""); response.Code == http.StatusOK { + t.Errorf("amd64 blob %s was mirrored despite the linux/arm64 filter", layer) + } + } + + registry.offline.Store(false) + riscv, _ := ParsePlatform("linux/riscv64") + if _, err := h.MirrorImage(context.Background(), ImageRef{Name: "team/app", References: []string{"multi"}}, []Platform{riscv}); err == nil { + t.Error("a filter matching no index entry did not fail") + } +} + +func TestMirrorImageMovedTagMirrorsBothTargets(t *testing.T) { + registry, image := newFakeImageRegistry(t) + proxy := newMirrorTestProxy(t, registry) + h := NewContainerHandlerWithRegistry(proxy, "http://proxy.example", registry.server.URL) + + result, err := h.MirrorImage(context.Background(), ImageRef{Name: "team/app", References: []string{image.singleDigest, "multi"}}, nil) + if err != nil { + t.Fatalf("MirrorImage: %v", err) + } + if !strings.Contains(strings.Join(result.Warnings, "\n"), "now points to") { + t.Errorf("warnings = %v, want the moved-tag warning", result.Warnings) + } + registry.offline.Store(true) + for _, layer := range []string{image.singleLayer, image.amd64Layer} { + if response := pull(t, h, "/team/app/blobs/"+layer, ""); response.Code != http.StatusOK { + t.Errorf("blob %s status = %d, want 200", layer, response.Code) + } + } +} + +func TestMirrorImageNamespaceRouteMatchesContainerdPull(t *testing.T) { + registry, image := newFakeImageRegistry(t) + proxy := newMirrorTestProxy(t, registry) + named := map[string]string{"internal": registry.server.URL} + h := NewContainerHandlerWithRegistry(proxy, "http://proxy.example", dockerHubRegistry, named) + host := strings.TrimPrefix(registry.server.URL, "http://") + + if _, err := h.MirrorImage(context.Background(), ImageRef{Registry: host, Name: "team/app", References: []string{"single"}}, nil); err != nil { + t.Fatalf("MirrorImage: %v", err) + } + registry.offline.Store(true) + ns := "?" + url.Values{namespaceQueryParam: {host}}.Encode() + if response := pull(t, h, "/team/app/manifests/single"+ns, containerdManifestAccept); response.Code != http.StatusOK { + t.Errorf("containerd ns manifest status = %d: %s", response.Code, response.Body.String()) + } + if response := pull(t, h, "/team/app/blobs/"+image.singleLayer+ns, ""); response.Code != http.StatusOK { + t.Errorf("containerd ns blob status = %d", response.Code) + } + if response := pull(t, h, "/upstream/internal/team/app/blobs/"+image.singleLayer, ""); response.Code != http.StatusOK { + t.Errorf("prefix-route blob status = %d", response.Code) + } + + if _, err := h.MirrorImage(context.Background(), ImageRef{Registry: "unknown.example", Name: "team/app", References: []string{"single"}}, nil); err == nil { + t.Error("an unconfigured registry host was accepted") + } +} + +func TestNormalizeImageRef(t *testing.T) { + tests := []struct { + name string + ref ImageRef + defaultRT string + wantName string + wantErr bool + }{ + {name: "docker hub default adds library", ref: ImageRef{Name: "alpine", References: []string{"3.20"}}, defaultRT: dockerHubRegistry, wantName: "library/alpine"}, + {name: "docker hub ns adds library", ref: ImageRef{Registry: "hub.docker.com", Name: "alpine", References: []string{"3.20"}}, defaultRT: "https://registry.example", wantName: "library/alpine"}, + {name: "other default keeps name", ref: ImageRef{Name: "alpine", References: []string{"3.20"}}, defaultRT: "https://registry.example", wantName: "alpine"}, + {name: "path traversal", ref: ImageRef{Name: "../../x", References: []string{"1"}}, defaultRT: dockerHubRegistry, wantErr: true}, + {name: "query in tag", ref: ImageRef{Name: "library/alpine", References: []string{"3.20?x#y"}}, defaultRT: dockerHubRegistry, wantErr: true}, + {name: "short digest", ref: ImageRef{Name: "library/alpine", References: []string{"sha256:abcd"}}, defaultRT: dockerHubRegistry, wantErr: true}, + {name: "no reference", ref: ImageRef{Name: "library/alpine"}, defaultRT: dockerHubRegistry, wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := normalizeImageRef(tt.ref, tt.defaultRT) + if (err != nil) != tt.wantErr { + t.Fatalf("err = %v, wantErr %v", err, tt.wantErr) + } + if !tt.wantErr && got.Name != tt.wantName { + t.Errorf("name = %q, want %q", got.Name, tt.wantName) + } + }) + } +} diff --git a/internal/mirror/image.go b/internal/mirror/image.go new file mode 100644 index 00000000..bc219372 --- /dev/null +++ b/internal/mirror/image.go @@ -0,0 +1,126 @@ +package mirror + +import ( + "context" + "errors" + "fmt" + "strings" + + "github.com/git-pkgs/proxy/internal/handler" +) + +// ImageMirrorer fetches container images into the cache. +type ImageMirrorer interface { + MirrorImage(ctx context.Context, ref handler.ImageRef, platforms []handler.Platform) (handler.ImageResult, error) +} + +// UnavailableImages refuses every image with the given reason. +type UnavailableImages struct { + Reason string +} + +func (u UnavailableImages) MirrorImage(context.Context, handler.ImageRef, []handler.Platform) (handler.ImageResult, error) { + return handler.ImageResult{}, errors.New(u.Reason) +} + +func isImageEcosystem(ecosystem string) bool { + return ecosystem == "docker" || ecosystem == "oci" +} + +// ParsePlatforms parses os/architecture[/variant] values. +func ParsePlatforms(values []string) ([]handler.Platform, error) { + var platforms []handler.Platform + for _, value := range values { + for _, item := range strings.Split(value, ",") { + item = strings.TrimSpace(item) + if item == "" { + continue + } + platform, err := handler.ParsePlatform(item) + if err != nil { + return nil, err + } + platforms = append(platforms, platform) + } + } + return platforms, nil +} + +// imageRef maps a docker or oci PURL to the image reference and platforms to +// mirror. +// +// - pkg:docker//@, optionally with +// repository_url=[/]. +// - pkg:oci/@?repository_url=/, optionally +// with tag= and arch=. OCI PURLs have no default +// registry, so repository_url is required. +func imageRef(pv PackageVersion) (handler.ImageRef, []handler.Platform, error) { + platforms, err := ParsePlatforms(pv.Platforms) + if err != nil { + return handler.ImageRef{}, nil, err + } + ref := handler.ImageRef{Name: pv.Name} + if pv.Version != "" { + ref.References = []string{pv.Version} + } + repositoryURL := pv.Qualifiers["repository_url"] + if pv.Ecosystem == "oci" { + if repositoryURL == "" { + return ref, nil, errors.New("pkg:oci needs a repository_url qualifier") + } + if tag := pv.Qualifiers["tag"]; tag != "" && tag != pv.Version { + ref.References = append(ref.References, tag) + } + if arch := pv.Qualifiers["arch"]; arch != "" && len(platforms) == 0 { + platforms = []handler.Platform{{Architecture: arch}} + } + } + if repositoryURL != "" { + repositoryURL = strings.TrimPrefix(strings.TrimPrefix(repositoryURL, "https://"), "http://") + host, repository, _ := strings.Cut(strings.Trim(repositoryURL, "/"), "/") + ref.Registry = host + switch { + case pv.Ecosystem == "oci" && repository != "": + // The repository URL names the repository itself. + ref.Name = repository + case repository != "": + ref.Name = repository + "/" + pv.Name + } + } + return ref, platforms, nil +} + +func (m *Mirror) mirrorImage(ctx context.Context, pv PackageVersion, tracker *progressTracker) { + fail := func(err error) { + tracker.failed.Add(1) + tracker.addError(pv.Ecosystem, pv.Name, pv.Version, err.Error()) + m.logger.Warn("mirror failed", "ecosystem", pv.Ecosystem, "name", pv.Name, "version", pv.Version, "error", err) + } + // pv.String keeps the namespace separator, which MakeString escapes for + // docker names. + if m.proxy != nil && m.proxy.Denylist.Denied(pv.String()) { + fail(fmt.Errorf("%w: %s", handler.ErrVersionDenied, pv)) + return + } + ref, platforms, err := imageRef(pv) + if err != nil { + fail(err) + return + } + result, err := m.Images.MirrorImage(ctx, ref, platforms) + for _, warning := range result.Warnings { + m.logger.Warn("mirror warning", "ecosystem", pv.Ecosystem, "name", pv.Name, "version", pv.Version, "warning", warning) + } + tracker.bytes.Add(result.Bytes) + if err != nil { + fail(err) + return + } + if result.Fetched == 0 { + tracker.skipped.Add(1) + m.logger.Debug("already cached", "ecosystem", pv.Ecosystem, "name", pv.Name, "version", pv.Version) + return + } + tracker.completed.Add(1) + m.logger.Info("mirrored", "ecosystem", pv.Ecosystem, "name", pv.Name, "version", pv.Version, "size", result.Bytes) +} diff --git a/internal/mirror/image_test.go b/internal/mirror/image_test.go new file mode 100644 index 00000000..339e86c2 --- /dev/null +++ b/internal/mirror/image_test.go @@ -0,0 +1,174 @@ +package mirror + +import ( + "context" + "errors" + "reflect" + "strings" + "sync" + "testing" + + "github.com/git-pkgs/proxy/internal/denylist" + "github.com/git-pkgs/proxy/internal/handler" +) + +func TestImageRefFromPURL(t *testing.T) { + const digest = "sha256:1111111111111111111111111111111111111111111111111111111111111111" + tests := []struct { + purl string + platforms []string + wantRef handler.ImageRef + wantPlatforms []handler.Platform + wantErr string + }{ + { + purl: "pkg:docker/library/alpine@3.20", + wantRef: handler.ImageRef{Name: "library/alpine", References: []string{"3.20"}}, + }, + { + purl: "pkg:docker/alpine@3.20?repository_url=ghcr.io", + platforms: []string{"linux/arm64/v8"}, + wantRef: handler.ImageRef{Registry: "ghcr.io", Name: "alpine", References: []string{"3.20"}}, + wantPlatforms: []handler.Platform{{OS: "linux", Architecture: "arm64", Variant: "v8"}}, + }, + { + purl: "pkg:docker/app@1.0?repository_url=https://registry.example/team", + wantRef: handler.ImageRef{Registry: "registry.example", Name: "team/app", References: []string{"1.0"}}, + }, + { + purl: "pkg:oci/debian@" + digest + "?repository_url=docker.io/library/debian&tag=bookworm&arch=arm64", + wantRef: handler.ImageRef{Registry: "docker.io", Name: "library/debian", References: []string{digest, "bookworm"}}, + wantPlatforms: []handler.Platform{{Architecture: "arm64"}}, + }, + { + purl: "pkg:oci/debian@" + digest + "?repository_url=docker.io/library/debian&arch=arm64", + platforms: []string{"linux/amd64"}, + wantRef: handler.ImageRef{Registry: "docker.io", Name: "library/debian", References: []string{digest}}, + wantPlatforms: []handler.Platform{{OS: "linux", Architecture: "amd64"}}, + }, + { + purl: "pkg:oci/debian@" + digest, + wantErr: "repository_url", + }, + } + for _, tt := range tests { + t.Run(tt.purl, func(t *testing.T) { + var items []PackageVersion + source := &PURLSource{PURLs: []string{tt.purl}, Platforms: tt.platforms} + if err := source.Enumerate(context.Background(), func(pv PackageVersion) error { + items = append(items, pv) + return nil + }); err != nil { + t.Fatalf("Enumerate: %v", err) + } + if len(items) != 1 { + t.Fatalf("items = %d, want 1", len(items)) + } + ref, platforms, err := imageRef(items[0]) + if tt.wantErr != "" { + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("err = %v, want %q", err, tt.wantErr) + } + return + } + if err != nil { + t.Fatalf("imageRef: %v", err) + } + if !reflect.DeepEqual(ref, tt.wantRef) { + t.Errorf("ref = %+v, want %+v", ref, tt.wantRef) + } + if !reflect.DeepEqual(platforms, tt.wantPlatforms) { + t.Errorf("platforms = %+v, want %+v", platforms, tt.wantPlatforms) + } + }) + } +} + +func TestUnversionedImagePURLIsNotExpanded(t *testing.T) { + var items []PackageVersion + source := &PURLSource{PURLs: []string{"pkg:docker/library/alpine"}} + if err := source.Enumerate(context.Background(), func(pv PackageVersion) error { + items = append(items, pv) + return nil + }); err != nil { + t.Fatalf("Enumerate: %v", err) + } + if len(items) != 1 || items[0].Version != "" { + t.Fatalf("items = %+v, want one unversioned item", items) + } +} + +type fakeImages struct { + mu sync.Mutex + refs []handler.ImageRef + result handler.ImageResult + err error +} + +func (f *fakeImages) MirrorImage(_ context.Context, ref handler.ImageRef, _ []handler.Platform) (handler.ImageResult, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.refs = append(f.refs, ref) + return f.result, f.err +} + +func TestMirrorRunCountsImages(t *testing.T) { + tests := []struct { + name string + result handler.ImageResult + err error + wantCompleted int64 + wantSkipped int64 + wantFailed int64 + }{ + {name: "fetched", result: handler.ImageResult{Fetched: 3, Bytes: 42}, wantCompleted: 1}, + {name: "cached", result: handler.ImageResult{}, wantSkipped: 1}, + {name: "failed", err: errors.New("boom"), wantFailed: 1}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + m := setupTestMirror(t, 1) + images := &fakeImages{result: tt.result, err: tt.err} + m.Images = images + progress, err := m.Run(context.Background(), &PURLSource{PURLs: []string{"pkg:docker/library/alpine@3.20"}}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if progress.Completed != tt.wantCompleted || progress.Skipped != tt.wantSkipped || progress.Failed != tt.wantFailed { + t.Errorf("progress = %+v", progress) + } + if progress.Bytes != tt.result.Bytes { + t.Errorf("bytes = %d, want %d", progress.Bytes, tt.result.Bytes) + } + if len(images.refs) != 1 || images.refs[0].Name != "library/alpine" { + t.Errorf("refs = %+v", images.refs) + } + }) + } +} + +func TestMirrorRunDeniedImage(t *testing.T) { + m := setupTestMirror(t, 1) + policy, err := denylist.New([]string{"pkg:docker/library/alpine@3.20"}) + if err != nil { + t.Fatal(err) + } + m.proxy.Denylist = policy + images := &fakeImages{} + m.Images = images + + progress, err := m.Run(context.Background(), &PURLSource{PURLs: []string{"pkg:docker/library/alpine@3.20"}}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if progress.Failed != 1 || len(images.refs) != 0 { + t.Errorf("denied image: failed = %d, mirror calls = %d", progress.Failed, len(images.refs)) + } +} + +func TestJobRequestRejectsInvalidPlatform(t *testing.T) { + js := &JobStore{} + if _, err := js.sourceFromRequest(JobRequest{PURLs: []string{"pkg:docker/library/alpine@3.20"}, Platforms: []string{"linux"}}); err == nil { + t.Error("invalid platform was accepted") + } +} diff --git a/internal/mirror/job.go b/internal/mirror/job.go index 70d70031..3d829667 100644 --- a/internal/mirror/job.go +++ b/internal/mirror/job.go @@ -39,6 +39,9 @@ type JobRequest struct { PURLs []string `json:"purls,omitempty"` SBOM json.RawMessage `json:"sbom,omitempty"` Registry string `json:"registry,omitempty"` + // Platforms limit which platforms of container images are mirrored, + // as os/architecture[/variant]. + Platforms []string `json:"platforms,omitempty"` } // JobStore manages in-memory mirror jobs. @@ -191,13 +194,16 @@ func (js *JobStore) runJob(ctx context.Context, cancel context.CancelFunc, job * } func (js *JobStore) sourceFromRequest(req JobRequest) (Source, error) { //nolint:ireturn // interface return is the design + if _, err := ParsePlatforms(req.Platforms); err != nil { + return nil, err + } switch { case len(req.PURLs) > 0 && len(req.SBOM) > 0: return nil, fmt.Errorf("request must include only one of purls or sbom") case len(req.PURLs) > 0: - return &PURLSource{PURLs: req.PURLs}, nil + return &PURLSource{PURLs: req.PURLs, Platforms: req.Platforms}, nil case len(req.SBOM) > 0: - return &SBOMSource{Data: req.SBOM}, nil + return &SBOMSource{Data: req.SBOM, Platforms: req.Platforms}, nil case req.Registry != "": return nil, fmt.Errorf("registry mirroring is not yet implemented; use purls instead") default: diff --git a/internal/mirror/mirror.go b/internal/mirror/mirror.go index b52c4f6a..fb98c11f 100644 --- a/internal/mirror/mirror.go +++ b/internal/mirror/mirror.go @@ -22,6 +22,10 @@ type Mirror struct { storage storage.Storage logger *slog.Logger workers int + + // Images mirrors docker and oci items. Without it they fail like any + // other unsupported ecosystem. + Images ImageMirrorer } // New creates a new Mirror with the given dependencies. @@ -203,6 +207,10 @@ func (m *Mirror) RunDryRun(ctx context.Context, source Source) ([]PackageVersion } func (m *Mirror) mirrorOne(ctx context.Context, pv PackageVersion, tracker *progressTracker) { + if isImageEcosystem(pv.Ecosystem) && m.Images != nil { + m.mirrorImage(ctx, pv, tracker) + return + } result, err := m.proxy.GetOrFetchArtifact(ctx, pv.Ecosystem, pv.Name, pv.Version, "") if err != nil { tracker.failed.Add(1) diff --git a/internal/mirror/source.go b/internal/mirror/source.go index 4269264c..6d3d096f 100644 --- a/internal/mirror/source.go +++ b/internal/mirror/source.go @@ -20,6 +20,11 @@ type PackageVersion struct { Ecosystem string Name string Version string + // Qualifiers are the PURL qualifiers. Images use repository_url, tag and + // arch. + Qualifiers map[string]string + // Platforms limit which platforms of an image index are mirrored. + Platforms []string } func (pv PackageVersion) String() string { @@ -36,6 +41,8 @@ type Source interface { type PURLSource struct { PURLs []string RegClient *registries.Client + // Platforms limit which platforms of container images are mirrored. + Platforms []string } func (s *PURLSource) Enumerate(ctx context.Context, fn func(PackageVersion) error) error { @@ -56,8 +63,15 @@ func (s *PURLSource) Enumerate(ctx context.Context, fn func(PackageVersion) erro name = p.Namespace + "/" + p.Name } - if p.Version != "" { - if err := fn(PackageVersion{Ecosystem: ecosystem, Name: name, Version: p.Version}); err != nil { + if p.Version != "" || isImageEcosystem(ecosystem) { + // Images are never expanded to all tags; an unversioned image + // fails when it is mirrored. + pv := PackageVersion{Ecosystem: ecosystem, Name: name, Version: p.Version} + if isImageEcosystem(ecosystem) { + pv.Qualifiers = p.Qualifiers.Map() + pv.Platforms = s.Platforms + } + if err := fn(pv); err != nil { return err } continue @@ -98,6 +112,8 @@ type SBOMSource struct { Data []byte Name string RegClient *registries.Client + // Platforms limit which platforms of container images are mirrored. + Platforms []string } func (s *SBOMSource) Enumerate(ctx context.Context, fn func(PackageVersion) error) error { @@ -109,7 +125,7 @@ func (s *SBOMSource) Enumerate(ctx context.Context, fn func(PackageVersion) erro return fmt.Errorf("parsing SBOM: %w", err) } - inner := &PURLSource{PURLs: purls, RegClient: s.RegClient} + inner := &PURLSource{PURLs: purls, RegClient: s.RegClient, Platforms: s.Platforms} return inner.Enumerate(ctx, fn) } diff --git a/internal/server/mirror_proxy_test.go b/internal/server/mirror_proxy_test.go index 2df395af..129a9131 100644 --- a/internal/server/mirror_proxy_test.go +++ b/internal/server/mirror_proxy_test.go @@ -40,7 +40,7 @@ func TestNewMirrorProxyFollowsBearerChallenge(t *testing.T) { registry := newTokenRegistry(t) cfg := config.Default() cfg.Upstream.AllowLoopback = true - proxy := NewMirrorProxy(cfg, nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) + proxy, _ := NewMirrorProxy(cfg, nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) resp, err := proxy.HTTPClient.Get(registry.URL + "/v2/library/app/manifests/1.0") if err != nil { @@ -54,7 +54,7 @@ func TestNewMirrorProxyFollowsBearerChallenge(t *testing.T) { func TestNewMirrorProxyRefusesLoopbackByDefault(t *testing.T) { registry := newTokenRegistry(t) - proxy := NewMirrorProxy(config.Default(), nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) + proxy, _ := NewMirrorProxy(config.Default(), nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) resp, err := proxy.HTTPClient.Get(registry.URL + "/v2/library/app/manifests/1.0") if err == nil { diff --git a/internal/server/server.go b/internal/server/server.go index de6204c1..7391c917 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -338,6 +338,7 @@ func (s *Server) serve(listener net.Listener) error { // Mirror API endpoints (opt-in via mirror_api config or PROXY_MIRROR_API env) if s.cfg.MirrorAPI { mirrorSvc := mirror.New(proxy, s.db, s.storage, s.logger, 4) //nolint:mnd // default concurrency + mirrorSvc.Images = s.newContainerHandler(proxy) jobStore := mirror.NewJobStore(bgCtx, mirrorSvc) mirrorAPI := NewMirrorAPIHandler(jobStore) r.Post("/api/mirror", mirrorAPI.HandleCreate) @@ -421,14 +422,7 @@ func (s *Server) mountProtocolHandlers(r chi.Router, proxy *handler.Proxy) { juliaHandler := handler.NewJuliaHandlerWithUpstream(proxy, s.cfg.Upstream.Julia) swiftHandler := handler.NewSwiftHandler(proxy, s.cfg.BaseURL, s.cfg.Upstream.Swift) homebrewHandler := handler.NewHomebrewHandler(proxy, s.cfg.Upstream.HomebrewAPI) - containerHandler := handler.NewContainerHandlerWithRegistry( - proxy, - s.cfg.BaseURL, - s.cfg.Upstream.OCIDefault, - s.cfg.Upstream.OCI, - ) - containerHandler.SetMirroredRegistries(s.cfg.Upstream.OCIMirrors) - handler.RegisterHomebrewArtifacts(containerHandler, s.cfg.Upstream.HomebrewArtifact) + containerHandler := s.newContainerHandler(proxy) helmHandler := handler.NewHelmHandlerWithOCIRegistries( proxy, s.cfg.BaseURL, s.cfg.Upstream.Helm, s.cfg.Upstream.OCIDefault, s.cfg.Upstream.OCI) apkHandler := handler.NewAPKHandler(proxy, s.cfg.BaseURL, s.cfg.Upstream.APK) @@ -488,15 +482,30 @@ func (s *Server) newUpstreamClients(artifactTimeout time.Duration) (metadataClie } // NewMirrorProxy builds the proxy for the mirror command with the same -// upstream clients, SSRF guard and upstream authentication as serve. Artifact +// upstream clients, SSRF guard and upstream authentication as serve, and a +// container handler with the same OCI routes for mirroring images. Artifact // downloads have no overall time limit; the caller cancels through ctx. -func NewMirrorProxy(cfg *config.Config, db *database.DB, store storage.Storage, logger *slog.Logger) *handler.Proxy { +func NewMirrorProxy(cfg *config.Config, db *database.DB, store storage.Storage, logger *slog.Logger) (*handler.Proxy, *handler.ContainerHandler) { s := &Server{cfg: cfg, db: db, storage: store, logger: logger, ecr: newECRTokens(logger)} metadataClient, artifactClient := s.newUpstreamClients(0) proxy := handler.NewProxy(db, store, fetch.NewFetcher(fetch.WithHTTPClient(artifactClient)), fetch.NewResolver(), logger) proxy.HTTPClient = metadataClient proxy.AuthForURL = s.authForURL - return proxy + return proxy, s.newContainerHandler(proxy) +} + +// newContainerHandler builds the OCI handler with the configured default +// registry, named upstreams, mirrored registries and Homebrew routes. +func (s *Server) newContainerHandler(proxy *handler.Proxy) *handler.ContainerHandler { + containerHandler := handler.NewContainerHandlerWithRegistry( + proxy, + s.cfg.BaseURL, + s.cfg.Upstream.OCIDefault, + s.cfg.Upstream.OCI, + ) + containerHandler.SetMirroredRegistries(s.cfg.Upstream.OCIMirrors) + handler.RegisterHomebrewArtifacts(containerHandler, s.cfg.Upstream.HomebrewArtifact) + return containerHandler } func configureScanning(proxy *handler.Proxy, cfg config.ScanningConfig, baseURL string, logger *slog.Logger) (*scanner.Group, error) {