From 055540a540614aa7d01dc550ca5cab451211b77b Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Sun, 11 Oct 2026 23:59:29 +0200 Subject: [PATCH 1/6] 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/6] 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/6] 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) { From 5b9a5fca6bf9441f00725d53d460dad7470c52a5 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Mon, 12 Oct 2026 10:29:12 +0200 Subject: [PATCH 4/6] When the shared tag entry holds a representation the client does not accept, the lookup now goes on to the client's own per-Accept entry instead of giving up, so a Docker-only client still gets its cached manifest list after an OCI index was cached for the same tag. --- internal/handler/container_manifest.go | 22 +++++++------- .../handler/container_manifest_key_test.go | 29 +++++++++++++++++++ 2 files changed, 40 insertions(+), 11 deletions(-) diff --git a/internal/handler/container_manifest.go b/internal/handler/container_manifest.go index 4ff750ed..9d8921b5 100644 --- a/internal/handler/container_manifest.go +++ b/internal/handler/container_manifest.go @@ -191,12 +191,13 @@ func (h *ContainerHandler) loadContainerManifestForAccept(ctx context.Context, r h.proxy.Logger.Warn("failed to read cached container manifest", "error", err) return nil } - if cached != nil { - if containerManifestCacheCompatible(accept, cached) { - return cached - } - return nil + if cached != nil && containerManifestCacheCompatible(accept, cached) { + return cached } + // The shared entry may hold a representation this client does not accept, + // for example an OCI index for a client that only accepts Docker manifest + // lists. Its own per-Accept entry is still written, so look there. + sharedEntryExists := cached != nil for _, legacyCacheKey := range h.containerManifestLegacyCacheKeys(registryURL, name, reference, accept, cacheAccept, cacheKey) { cached, err = h.loadContainerManifest(ctx, legacyCacheKey) @@ -204,14 +205,13 @@ func (h *ContainerHandler) loadContainerManifestForAccept(ctx context.Context, r h.proxy.Logger.Warn("failed to read legacy cached container manifest", "error", err) return nil } - if cached == nil { + if cached == nil || !containerManifestCacheCompatible(accept, cached) { 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) + if !sharedEntryExists { + if err := h.storeContainerManifest(ctx, cacheKey, cached); err != nil { + h.proxy.Logger.Warn("failed to migrate cached container manifest", "error", err) + } } return cached } diff --git a/internal/handler/container_manifest_key_test.go b/internal/handler/container_manifest_key_test.go index e339ced0..62ceac8a 100644 --- a/internal/handler/container_manifest_key_test.go +++ b/internal/handler/container_manifest_key_test.go @@ -163,3 +163,32 @@ func TestContainerHandler_IndexlessClientKeepsOwnTagEntry(t *testing.T) { t.Errorf("offline index-less body = %q, want the converted manifest", response.Body.String()) } } + +func TestContainerHandler_IncompatibleSharedEntryFallsBackToAcceptEntry(t *testing.T) { + list := `{"schemaVersion":2,"mediaType":"application/vnd.docker.distribution.manifest.list.v2+json","manifests":[]}` + index := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.index.v1+json","manifests":[]}` + registry := newFakeManifestRegistry(t) + registry.set("/v2/library/app/manifests/latest", dockerManifestListMediaType, list) + h := newManifestKeyTestHandler(t, registry) + + const dockerOnly = "application/vnd.docker.distribution.manifest.list.v2+json, application/vnd.docker.distribution.manifest.v2+json" + if response := getManifest(t, h, "/library/app/manifests/latest", dockerOnly); response.Body.String() != list { + t.Fatalf("Docker-only pull body = %q, want the manifest list", response.Body.String()) + } + registry.set("/v2/library/app/manifests/latest", ociImageIndexMediaType, index) + if response := getManifest(t, h, "/library/app/manifests/latest", ociImageIndexMediaType); response.Body.String() != index { + t.Fatalf("OCI pull body = %q, want the index", response.Body.String()) + } + registry.offline.Store(true) + + response := getManifest(t, h, "/library/app/manifests/latest", dockerOnly) + if response.Code != http.StatusOK { + t.Fatalf("offline Docker-only pull status = %d, want 200: %s", response.Code, response.Body.String()) + } + if response.Body.String() != list { + t.Errorf("offline Docker-only body = %q, want the manifest list", response.Body.String()) + } + if response := getManifest(t, h, "/library/app/manifests/latest", ociImageIndexMediaType); response.Body.String() != index { + t.Errorf("offline OCI body = %q, want the index still in the shared entry", response.Body.String()) + } +} From 67fd1ad6c1ee128455cd7a770c8e1f45c0a4d024 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Mon, 12 Oct 2026 10:36:51 +0200 Subject: [PATCH 5/6] The mirror now resolves a tag without caching it, mirrors the image it points to, and stores the tag only once that image is complete, so a failed or interrupted mirror keeps the previously cached tag instead of pointing it at an image with missing layers. --- internal/handler/container_mirror.go | 87 ++++++++++++++++++++++- internal/handler/container_mirror_test.go | 38 ++++++++++ 2 files changed, 124 insertions(+), 1 deletion(-) diff --git a/internal/handler/container_mirror.go b/internal/handler/container_mirror.go index f9e3d495..aa0c7644 100644 --- a/internal/handler/container_mirror.go +++ b/internal/handler/container_mirror.go @@ -11,6 +11,7 @@ import ( "regexp" "slices" "strings" + "time" ) const ( @@ -160,7 +161,11 @@ func (h *ContainerHandler) MirrorImage(ctx context.Context, ref ImageRef, platfo } var digests []string for _, reference := range ref.References { - digest, err := m.mirrorManifest(ctx, reference, 0) + mirror := m.mirrorTag + if mirrorDigestPattern.MatchString(reference) { + mirror = func(ctx context.Context, digest string) (string, error) { return m.mirrorManifest(ctx, digest, 0) } + } + digest, err := mirror(ctx, reference) if err != nil { return m.result, fmt.Errorf("%s:%s: %w", ref.Name, reference, err) } @@ -197,6 +202,86 @@ func normalizeImageRef(ref ImageRef, defaultRegistryURL string) (ImageRef, error return ref, nil } +// errTagUpstreamUnavailable marks tag lookups that failed because the +// upstream could not be reached or answered with 429 or 5xx. +var errTagUpstreamUnavailable = errors.New("upstream unavailable") + +// mirrorTag resolves a tag upstream without caching it, mirrors the manifest +// it names by digest, and stores the tag only after that manifest and +// everything it references are cached. A failed mirror leaves the previously +// cached tag as it was. If the upstream is unreachable, the cached tag is +// mirrored as it is. +func (m *imageMirror) mirrorTag(ctx context.Context, tag string) (string, error) { + accept := containerManifestAccept(&http.Request{Header: http.Header{}}) + cacheAccept := normalizeContainerManifestAccept(accept) + before := m.h.loadContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, tag, accept, cacheAccept) + + manifest, err := m.fetchTagManifest(ctx, tag, accept) + if err != nil { + if before == nil || !errors.Is(err, errTagUpstreamUnavailable) { + return "", fmt.Errorf("manifest %s: %w", tag, err) + } + m.result.Warnings = append(m.result.Warnings, fmt.Sprintf("manifest %s served from stale cache, upstream unreachable", tag)) + _, err := m.mirrorManifest(ctx, before.contentDigest, 0) + return before.contentDigest, err + } + + // The digest entry is immutable, so it can be stored right away; the + // walk below then reads it from cache instead of fetching it again. + digest := manifest.contentDigest + if before == nil || before.contentDigest != digest { + m.result.Fetched++ + } + m.h.storeContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, digest, accept, cacheAccept, manifest) + if _, err := m.mirrorManifest(ctx, digest, 0); err != nil { + return "", err + } + m.h.storeContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, tag, accept, cacheAccept, manifest) + return digest, nil +} + +// fetchTagManifest fetches and verifies the manifest a tag points to without +// caching it. +func (m *imageMirror) fetchTagManifest(ctx context.Context, tag, accept string) (*cachedContainerManifest, error) { + upstreamURL := fmt.Sprintf("%s/v2/%s/manifests/%s", m.registryURL, m.upstreamName, tag) + request, err := http.NewRequestWithContext(ctx, http.MethodGet, upstreamURL, nil) + if err != nil { + return nil, err + } + request.Header.Set("Accept", accept) + response, err := m.h.proxy.HTTPClient.Do(request) + if err != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } + return nil, fmt.Errorf("%w: %w", errTagUpstreamUnavailable, err) + } + defer func() { _ = response.Body.Close() }() + if response.StatusCode != http.StatusOK { + if shouldServeStaleManifest(response.StatusCode) { + return nil, fmt.Errorf("%w: status %d", errTagUpstreamUnavailable, response.StatusCode) + } + return nil, fmt.Errorf("upstream status %d", response.StatusCode) + } + body, err := m.h.proxy.ReadMetadata(response.Body) + if err != nil { + return nil, fmt.Errorf("reading manifest: %w", err) + } + digest := sha256Digest(body) + if contentDigest := response.Header.Get("Docker-Content-Digest"); strings.HasPrefix(contentDigest, "sha256:") && contentDigest != digest { + return nil, fmt.Errorf("digest verification failed: upstream says %s, body is %s", contentDigest, digest) + } + return &cachedContainerManifest{ + body: body, + contentType: response.Header.Get(headerContentType), + contentDigest: digest, + etag: response.Header.Get(headerETag), + size: int64(len(body)), + lastModified: parseHTTPTime(response.Header.Get(headerLastModified)), + fetchedAt: time.Now(), + }, nil +} + // mirrorManifest stores the manifest for reference through the handler's own // manifest route, then follows what it references. It returns the manifest // digest. diff --git a/internal/handler/container_mirror_test.go b/internal/handler/container_mirror_test.go index fc4e8a60..ebd32b6d 100644 --- a/internal/handler/container_mirror_test.go +++ b/internal/handler/container_mirror_test.go @@ -293,3 +293,41 @@ func TestNormalizeImageRef(t *testing.T) { }) } } + +func TestMirrorImageKeepsPreviousTagWhenMovedTagIsIncomplete(t *testing.T) { + registry, image := newFakeImageRegistry(t) + proxy := newMirrorTestProxy(t, registry) + h := NewContainerHandlerWithRegistry(proxy, "http://proxy.example", registry.server.URL) + ref := ImageRef{Name: "team/app", References: []string{"single"}} + + if _, err := h.MirrorImage(context.Background(), ref, nil); err != nil { + t.Fatalf("first MirrorImage: %v", err) + } + previous := registry.manifests["/v2/team/app/manifests/single"] + + // Move the tag to an image whose layer the registry cannot serve. + broken, err := json.Marshal(map[string]any{ + "schemaVersion": 2, "mediaType": ociManifestMediaType, + "config": map[string]any{"mediaType": "application/vnd.oci.image.config.v1+json", "digest": image.singleLayer}, + "layers": []map[string]any{{"mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", "digest": sha256Digest([]byte("missing layer"))}}, + }) + if err != nil { + t.Fatal(err) + } + brokenDoc := fakeManifest{contentType: ociManifestMediaType, body: string(broken)} + registry.manifests["/v2/team/app/manifests/single"] = brokenDoc + registry.manifests["/v2/team/app/manifests/"+sha256Digest(broken)] = brokenDoc + + if _, err := h.MirrorImage(context.Background(), ref, nil); err == nil { + t.Fatal("mirroring the incomplete image did not fail") + } + registry.offline.Store(true) + + response := pull(t, h, "/team/app/manifests/single", dockerManifestAccept) + if response.Code != http.StatusOK { + t.Fatalf("offline tag status = %d: %s", response.Code, response.Body.String()) + } + if response.Body.String() != previous.body { + t.Errorf("offline tag points to %s, want the previously mirrored image", sha256Digest(response.Body.Bytes())) + } +} From b0570bdf9d5d88afc33f7854a1967c47916f3006 Mon Sep 17 00:00:00 2001 From: Christian Heim Date: Tue, 13 Oct 2026 17:02:32 +0200 Subject: [PATCH 6/6] Mirroring a tag now checks it with a HEAD first and reuses the cached manifest when the digest is unchanged, falls back to the cached tag when the upstream body is cut off, and keeps a tag that a client pull moved while the walk ran. The platform filter is checked across the whole image, so a nested index without the wanted platform no longer fails it, and a 304 for a per-Accept fallback entry only refreshes that client's own keys instead of overwriting the shared entry. --- internal/handler/container_manifest.go | 31 +++-- .../handler/container_manifest_key_test.go | 53 +++++++- internal/handler/container_mirror.go | 110 +++++++++++----- internal/handler/container_mirror_test.go | 123 +++++++++++++++++- 4 files changed, 276 insertions(+), 41 deletions(-) diff --git a/internal/handler/container_manifest.go b/internal/handler/container_manifest.go index 9d8921b5..cb390bd1 100644 --- a/internal/handler/container_manifest.go +++ b/internal/handler/container_manifest.go @@ -47,7 +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) - cached := h.loadContainerManifestForAccept(r.Context(), registryURL, name, reference, accept, cacheAccept) + cached, fromFallback := h.loadContainerManifestForAccept(r.Context(), registryURL, name, reference, accept, cacheAccept) immutable := manifestDigestReferencePattern.MatchString(reference) if cached != nil && (immutable || h.containerManifestFresh(cached)) { @@ -75,7 +75,13 @@ func (h *ContainerHandler) serveManifest(w http.ResponseWriter, r *http.Request, if resp.StatusCode == http.StatusNotModified && cached != nil { cached.fetchedAt = time.Now() - h.storeContainerManifestForAccept(r.Context(), registryURL, name, reference, accept, cacheAccept, cached) + if fromFallback { + // The shared entry holds another representation for other + // clients; only this client's own entries are refreshed. + h.storeContainerManifestLegacyEntries(r.Context(), registryURL, name, reference, accept, cacheAccept, cached) + } else { + h.storeContainerManifestForAccept(r.Context(), registryURL, name, reference, accept, cacheAccept, cached) + } writeContainerManifest(w, r, cached, false) return } @@ -184,15 +190,18 @@ func (h *ContainerHandler) containerManifestLegacyCacheKeys(registryURL, name, r return keys } -func (h *ContainerHandler) loadContainerManifestForAccept(ctx context.Context, registryURL, name, reference, accept, cacheAccept string) *cachedContainerManifest { +// loadContainerManifestForAccept returns the cached manifest for the client and +// whether it came from a per-Accept fallback key while the shared key holds a +// representation the client does not accept. +func (h *ContainerHandler) loadContainerManifestForAccept(ctx context.Context, registryURL, name, reference, accept, cacheAccept string) (*cachedContainerManifest, bool) { 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) - return nil + return nil, false } if cached != nil && containerManifestCacheCompatible(accept, cached) { - return cached + return cached, false } // The shared entry may hold a representation this client does not accept, // for example an OCI index for a client that only accepts Docker manifest @@ -203,7 +212,7 @@ func (h *ContainerHandler) loadContainerManifestForAccept(ctx context.Context, r cached, err = h.loadContainerManifest(ctx, legacyCacheKey) if err != nil { h.proxy.Logger.Warn("failed to read legacy cached container manifest", "error", err) - return nil + return nil, false } if cached == nil || !containerManifestCacheCompatible(accept, cached) { continue @@ -213,9 +222,9 @@ func (h *ContainerHandler) loadContainerManifestForAccept(ctx context.Context, r h.proxy.Logger.Warn("failed to migrate cached container manifest", "error", err) } } - return cached + return cached, sharedEntryExists } - return nil + return nil, false } func (h *ContainerHandler) storeContainerManifestForAccept(ctx context.Context, registryURL, name, reference, accept, cacheAccept string, manifest *cachedContainerManifest) { @@ -224,6 +233,12 @@ func (h *ContainerHandler) storeContainerManifestForAccept(ctx context.Context, h.proxy.Logger.Warn("failed to cache container manifest", "error", err) } + h.storeContainerManifestLegacyEntries(ctx, registryURL, name, reference, accept, cacheAccept, manifest) +} + +// storeContainerManifestLegacyEntries writes only the per-Accept keys. +func (h *ContainerHandler) storeContainerManifestLegacyEntries(ctx context.Context, registryURL, name, reference, accept, cacheAccept string, manifest *cachedContainerManifest) { + cacheKey := h.containerManifestSharedCacheKey(registryURL, name, reference, cacheAccept) 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 index 62ceac8a..75ad3d57 100644 --- a/internal/handler/container_manifest_key_test.go +++ b/internal/handler/container_manifest_key_test.go @@ -4,6 +4,7 @@ import ( "io" "net/http" "net/http/httptest" + "strings" "sync/atomic" "testing" ) @@ -188,7 +189,55 @@ func TestContainerHandler_IncompatibleSharedEntryFallsBackToAcceptEntry(t *testi if response.Body.String() != list { t.Errorf("offline Docker-only body = %q, want the manifest list", response.Body.String()) } - if response := getManifest(t, h, "/library/app/manifests/latest", ociImageIndexMediaType); response.Body.String() != index { - t.Errorf("offline OCI body = %q, want the index still in the shared entry", response.Body.String()) + if response.Header().Get("Warning") == "" { + t.Error("offline Docker-only response has no stale warning, so it did not come from the stale path") + } + // Serving the fallback entry must not copy it over the shared entry, + // which still holds the index for clients that accept both. + if response := getManifest(t, h, "/library/app/manifests/latest", dockerManifestAccept); response.Body.String() != index { + t.Errorf("offline body for a client accepting both = %q, want the index still in the shared entry", response.Body.String()) + } +} + +func TestContainerHandler_NotModifiedFallbackKeepsSharedEntry(t *testing.T) { + list := `{"schemaVersion":2,"mediaType":"application/vnd.docker.distribution.manifest.list.v2+json","manifests":[]}` + index := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.index.v1+json","manifests":[]}` + var offline atomic.Bool + // The registry answers per Accept header, with ETags, like registries + // that convert OCI indexes for clients without OCI support. + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if offline.Load() { + http.Error(w, "offline", http.StatusServiceUnavailable) + return + } + body, contentType := list, dockerManifestListMediaType + if strings.Contains(r.Header.Get("Accept"), ociImageIndexMediaType) { + body, contentType = index, ociImageIndexMediaType + } + etag := `"` + sha256Digest([]byte(body)) + `"` + if r.Header.Get("If-None-Match") == etag { + w.WriteHeader(http.StatusNotModified) + return + } + w.Header().Set("Content-Type", contentType) + w.Header().Set("ETag", etag) + _, _ = io.WriteString(w, body) + })) + t.Cleanup(upstream.Close) + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = upstream.Client() + h := &ContainerHandler{proxy: proxy, registryURL: upstream.URL, proxyURL: "http://localhost:8080"} + + const dockerOnly = "application/vnd.docker.distribution.manifest.list.v2+json, application/vnd.docker.distribution.manifest.v2+json" + getManifest(t, h, "/library/app/manifests/latest", dockerOnly) + getManifest(t, h, "/library/app/manifests/latest", ociImageIndexMediaType) + // Revalidates the Docker-only fallback entry and gets 304. + if response := getManifest(t, h, "/library/app/manifests/latest", dockerOnly); response.Body.String() != list { + t.Fatalf("Docker-only revalidation body = %q, want the list", response.Body.String()) + } + offline.Store(true) + + if response := getManifest(t, h, "/library/app/manifests/latest", dockerManifestAccept); response.Body.String() != index { + t.Errorf("body for a client accepting both = %q, want the index; the 304 refresh overwrote the shared entry", response.Body.String()) } } diff --git a/internal/handler/container_mirror.go b/internal/handler/container_mirror.go index aa0c7644..cb98a971 100644 --- a/internal/handler/container_mirror.go +++ b/internal/handler/container_mirror.go @@ -127,6 +127,9 @@ type imageMirror struct { platforms []Platform seen map[string]bool result ImageResult + // platformMatches counts index entries that matched the platform filter + // during the current walk. + platformMatches int } // MirrorImage fetches an image the way a client pull does: the manifest or @@ -214,9 +217,9 @@ var errTagUpstreamUnavailable = errors.New("upstream unavailable") func (m *imageMirror) mirrorTag(ctx context.Context, tag string) (string, error) { accept := containerManifestAccept(&http.Request{Header: http.Header{}}) cacheAccept := normalizeContainerManifestAccept(accept) - before := m.h.loadContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, tag, accept, cacheAccept) + before, _ := m.h.loadContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, tag, accept, cacheAccept) - manifest, err := m.fetchTagManifest(ctx, tag, accept) + manifest, err := m.resolveTag(ctx, tag, accept, before) if err != nil { if before == nil || !errors.Is(err, errTagUpstreamUnavailable) { return "", fmt.Errorf("manifest %s: %w", tag, err) @@ -231,41 +234,53 @@ func (m *imageMirror) mirrorTag(ctx context.Context, tag string) (string, error) digest := manifest.contentDigest if before == nil || before.contentDigest != digest { m.result.Fetched++ + m.h.storeContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, digest, accept, cacheAccept, manifest) } - m.h.storeContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, digest, accept, cacheAccept, manifest) if _, err := m.mirrorManifest(ctx, digest, 0); err != nil { return "", err } + + // A client pull may have moved the tag while the walk ran. Its entry is + // newer than what this mirror resolved, so it is kept. + current, _ := m.h.loadContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, tag, accept, cacheAccept) + if current != nil && current.contentDigest != digest && (before == nil || current.contentDigest != before.contentDigest) { + m.result.Warnings = append(m.result.Warnings, fmt.Sprintf("tag %s moved to %s while it was mirrored; kept that entry", tag, current.contentDigest)) + return digest, nil + } m.h.storeContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, tag, accept, cacheAccept, manifest) return digest, nil } -// fetchTagManifest fetches and verifies the manifest a tag points to without -// caching it. -func (m *imageMirror) fetchTagManifest(ctx context.Context, tag, accept string) (*cachedContainerManifest, error) { - upstreamURL := fmt.Sprintf("%s/v2/%s/manifests/%s", m.registryURL, m.upstreamName, tag) - request, err := http.NewRequestWithContext(ctx, http.MethodGet, upstreamURL, nil) +// resolveTag returns the manifest a tag points to upstream without caching +// it. A HEAD request comes first: if the tag still points to the cached +// digest, the cached manifest is reused with a new fetch time, which spares a +// GET (registries such as Docker Hub do not count HEAD requests against pull +// limits). +func (m *imageMirror) resolveTag(ctx context.Context, tag, accept string, before *cachedContainerManifest) (*cachedContainerManifest, error) { + if before != nil { + response, err := m.requestTag(ctx, http.MethodHead, tag, accept) + if err != nil { + return nil, err + } + _ = response.Body.Close() + if response.Header.Get("Docker-Content-Digest") == before.contentDigest { + unchanged := *before + unchanged.fetchedAt = time.Now() + return &unchanged, nil + } + } + + response, err := m.requestTag(ctx, http.MethodGet, tag, accept) if err != nil { return nil, err } - request.Header.Set("Accept", accept) - response, err := m.h.proxy.HTTPClient.Do(request) + defer func() { _ = response.Body.Close() }() + body, err := m.h.proxy.ReadMetadata(response.Body) if err != nil { if ctx.Err() != nil { return nil, ctx.Err() } - return nil, fmt.Errorf("%w: %w", errTagUpstreamUnavailable, err) - } - defer func() { _ = response.Body.Close() }() - if response.StatusCode != http.StatusOK { - if shouldServeStaleManifest(response.StatusCode) { - return nil, fmt.Errorf("%w: status %d", errTagUpstreamUnavailable, response.StatusCode) - } - return nil, fmt.Errorf("upstream status %d", response.StatusCode) - } - body, err := m.h.proxy.ReadMetadata(response.Body) - if err != nil { - return nil, fmt.Errorf("reading manifest: %w", err) + return nil, fmt.Errorf("%w: reading manifest: %w", errTagUpstreamUnavailable, err) } digest := sha256Digest(body) if contentDigest := response.Header.Get("Docker-Content-Digest"); strings.HasPrefix(contentDigest, "sha256:") && contentDigest != digest { @@ -282,12 +297,38 @@ func (m *imageMirror) fetchTagManifest(ctx context.Context, tag, accept string) }, nil } +// requestTag sends a manifest request for a tag upstream. Only 200 responses +// are returned; network errors, 429 and 5xx wrap errTagUpstreamUnavailable. +func (m *imageMirror) requestTag(ctx context.Context, method, tag, accept string) (*http.Response, error) { + upstreamURL := fmt.Sprintf("%s/v2/%s/manifests/%s", m.registryURL, m.upstreamName, tag) + request, err := http.NewRequestWithContext(ctx, method, upstreamURL, nil) + if err != nil { + return nil, err + } + request.Header.Set("Accept", accept) + response, err := m.h.proxy.HTTPClient.Do(request) + if err != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } + return nil, fmt.Errorf("%w: %w", errTagUpstreamUnavailable, err) + } + if response.StatusCode != http.StatusOK { + _ = response.Body.Close() + if shouldServeStaleManifest(response.StatusCode) { + return nil, fmt.Errorf("%w: status %d", errTagUpstreamUnavailable, response.StatusCode) + } + return nil, fmt.Errorf("upstream status %d", response.StatusCode) + } + return response, 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) + before, _ := m.h.loadContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, reference, cacheAccept, cacheAccept) target := "/" + m.ref.Name + "/manifests/" + reference if m.ref.Registry != "" { @@ -324,10 +365,21 @@ func (m *imageMirror) mirrorManifest(ctx context.Context, reference string, dept 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) + if manifest.Manifests == nil { + return digest, m.mirrorBlobs(ctx, &manifest) } - return digest, m.mirrorBlobs(ctx, &manifest) + if depth == 0 { + m.platformMatches = 0 + } + if err := m.mirrorIndex(ctx, reference, &manifest, depth); err != nil { + return "", err + } + // A nested index may lack the wanted platforms as long as the image as a + // whole has them. + if depth == 0 && len(m.platforms) > 0 && m.platformMatches == 0 { + return "", fmt.Errorf("index %s: no manifest for the requested platforms", reference) + } + return digest, nil } // mirrorIndex follows the index entries for the wanted platforms. Docker @@ -350,11 +402,11 @@ func (m *imageMirror) mirrorIndex(ctx context.Context, reference string, index * if m.wantsPlatform(entry.Platform) { selected = append(selected, entry) selectedDigests[entry.Digest] = true + if entry.Platform != nil { + m.platformMatches++ + } } } - 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) diff --git a/internal/handler/container_mirror_test.go b/internal/handler/container_mirror_test.go index ebd32b6d..332851a6 100644 --- a/internal/handler/container_mirror_test.go +++ b/internal/handler/container_mirror_test.go @@ -7,6 +7,7 @@ import ( "net/http" "net/http/httptest" "net/url" + "strconv" "strings" "sync/atomic" "testing" @@ -22,6 +23,12 @@ type fakeImageRegistry struct { manifests map[string]fakeManifest blobs map[string]string blobGets atomic.Int32 + // tagGets counts GET requests for manifests by tag. + tagGets atomic.Int32 + // truncateManifests makes manifest GETs end after a few bytes. + truncateManifests atomic.Bool + // onBlob, if set, runs before a blob is served. + onBlob func(path string) } type fakeImage struct { @@ -112,10 +119,21 @@ func newFakeImageRegistry(t *testing.T) (*fakeImageRegistry, fakeImage) { 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))) + if req.Method == http.MethodGet && !strings.Contains(req.URL.Path, "/manifests/sha256:") { + r.tagGets.Add(1) + } + if req.Method == http.MethodGet && r.truncateManifests.Load() { + w.Header().Set("Content-Length", strconv.Itoa(len(doc.body))) + _, _ = io.WriteString(w, doc.body[:4]) + return + } _, _ = io.WriteString(w, doc.body) return } if blob, ok := r.blobs[req.URL.Path]; ok { + if r.onBlob != nil { + r.onBlob(req.URL.Path) + } r.blobGets.Add(1) _, _ = io.WriteString(w, blob) return @@ -318,8 +336,9 @@ func TestMirrorImageKeepsPreviousTagWhenMovedTagIsIncomplete(t *testing.T) { registry.manifests["/v2/team/app/manifests/single"] = brokenDoc registry.manifests["/v2/team/app/manifests/"+sha256Digest(broken)] = brokenDoc - if _, err := h.MirrorImage(context.Background(), ref, nil); err == nil { - t.Fatal("mirroring the incomplete image did not fail") + missing := sha256Digest([]byte("missing layer")) + if _, err := h.MirrorImage(context.Background(), ref, nil); err == nil || !strings.Contains(err.Error(), missing) { + t.Fatalf("mirroring the incomplete image: err = %v, want a failure naming layer %s", err, missing) } registry.offline.Store(true) @@ -331,3 +350,103 @@ func TestMirrorImageKeepsPreviousTagWhenMovedTagIsIncomplete(t *testing.T) { t.Errorf("offline tag points to %s, want the previously mirrored image", sha256Digest(response.Body.Bytes())) } } + +func TestMirrorImagePlatformFilterAcrossNestedIndex(t *testing.T) { + registry, image := newFakeImageRegistry(t) + proxy := newMirrorTestProxy(t, registry) + h := NewContainerHandlerWithRegistry(proxy, "http://proxy.example", registry.server.URL) + + nested := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.index.v1+json","manifests":[` + + `{"mediaType":"application/vnd.oci.image.manifest.v1+json","digest":"` + image.arm64Digest + `","platform":{"os":"linux","architecture":"arm64"}}]}` + nestedDigest := sha256Digest([]byte(nested)) + top := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.index.v1+json","manifests":[` + + `{"mediaType":"application/vnd.oci.image.manifest.v1+json","digest":"` + image.amd64Digest + `","platform":{"os":"linux","architecture":"amd64"}},` + + `{"mediaType":"application/vnd.oci.image.index.v1+json","digest":"` + nestedDigest + `"}]}` + for _, reference := range []string{"nested", nestedDigest} { + registry.manifests["/v2/team/app/manifests/"+reference] = fakeManifest{contentType: ociImageIndexMediaType, body: nested} + } + for _, reference := range []string{"top", sha256Digest([]byte(top))} { + registry.manifests["/v2/team/app/manifests/"+reference] = fakeManifest{contentType: ociImageIndexMediaType, body: top} + } + + amd64, _ := ParsePlatform("linux/amd64") + if _, err := h.MirrorImage(context.Background(), ImageRef{Name: "team/app", References: []string{"top"}}, []Platform{amd64}); err != nil { + t.Fatalf("MirrorImage: %v (a nested index without amd64 must not fail the image)", err) + } + riscv, _ := ParsePlatform("linux/riscv64") + if _, err := h.MirrorImage(context.Background(), ImageRef{Name: "team/app", References: []string{"top"}}, []Platform{riscv}); err == nil { + t.Error("a filter matching nothing in the whole image did not fail") + } +} + +func TestMirrorImageRevalidatesUnchangedTagWithHead(t *testing.T) { + registry, _ := newFakeImageRegistry(t) + proxy := newMirrorTestProxy(t, registry) + h := NewContainerHandlerWithRegistry(proxy, "http://proxy.example", registry.server.URL) + ref := ImageRef{Name: "team/app", References: []string{"multi"}} + + if _, err := h.MirrorImage(context.Background(), ref, nil); err != nil { + t.Fatalf("first MirrorImage: %v", err) + } + gets := registry.tagGets.Load() + if _, err := h.MirrorImage(context.Background(), ref, nil); err != nil { + t.Fatalf("second MirrorImage: %v", err) + } + if extra := registry.tagGets.Load() - gets; extra != 0 { + t.Errorf("second run sent %d GETs for the unchanged tag, want a HEAD only", extra) + } +} + +func TestMirrorImageFallsBackWhenTagBodyIsCutOff(t *testing.T) { + registry, image := newFakeImageRegistry(t) + proxy := newMirrorTestProxy(t, registry) + h := NewContainerHandlerWithRegistry(proxy, "http://proxy.example", registry.server.URL) + ref := ImageRef{Name: "team/app", References: []string{"single"}} + + if _, err := h.MirrorImage(context.Background(), ref, nil); err != nil { + t.Fatalf("first MirrorImage: %v", err) + } + // Point the tag elsewhere so the HEAD check sends a GET, then cut that + // GET's body short. + registry.manifests["/v2/team/app/manifests/single"] = registry.manifests["/v2/team/app/manifests/"+image.amd64Digest] + registry.truncateManifests.Store(true) + + result, err := h.MirrorImage(context.Background(), ref, nil) + if err != nil { + t.Fatalf("MirrorImage with a cut-off body: %v, want the cached tag to be used", err) + } + if !strings.Contains(strings.Join(result.Warnings, "\n"), "stale cache") { + t.Errorf("warnings = %v, want the stale-cache warning", result.Warnings) + } +} + +func TestMirrorImageKeepsTagMovedByPullDuringWalk(t *testing.T) { + registry, image := newFakeImageRegistry(t) + proxy := newMirrorTestProxy(t, registry) + h := NewContainerHandlerWithRegistry(proxy, "http://proxy.example", registry.server.URL) + + // While the mirror fetches the first layer of "single", the tag moves + // upstream and a client pull caches the new target. + moved := registry.manifests["/v2/team/app/manifests/"+image.amd64Digest] + var once atomic.Bool + registry.onBlob = func(string) { + if once.CompareAndSwap(false, true) { + registry.manifests["/v2/team/app/manifests/single"] = moved + if response := pull(t, h, "/team/app/manifests/single", dockerManifestAccept); response.Code != http.StatusOK { + t.Errorf("client pull during the walk: status %d", response.Code) + } + } + } + result, err := h.MirrorImage(context.Background(), ImageRef{Name: "team/app", References: []string{"single"}}, nil) + if err != nil { + t.Fatalf("MirrorImage: %v", err) + } + if !strings.Contains(strings.Join(result.Warnings, "\n"), "moved") { + t.Errorf("warnings = %v, want the moved-tag warning", result.Warnings) + } + registry.onBlob = nil + registry.offline.Store(true) + if response := pull(t, h, "/team/app/manifests/single", dockerManifestAccept); response.Body.String() != moved.body { + t.Errorf("offline tag points to %s, want the newer target %s", sha256Digest(response.Body.Bytes()), image.amd64Digest) + } +}