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 5b625631..b4208d15 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 @@ -421,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") @@ -428,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 @@ -445,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() @@ -518,10 +526,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, images := server.NewMirrorProxy(cfg, db, store, logger) proxy.Denylist, err = denylist.New(cfg.Denylist.Packages) if err != nil { _ = db.Close() @@ -533,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_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_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/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") } } 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 new file mode 100644 index 00000000..129a9131 --- /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..7391c917 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 { @@ -350,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) @@ -433,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) @@ -481,6 +463,51 @@ 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, 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, *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, 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) { scanGroup, err := scanner.NewGroup(cfg, logger) if err != nil {