diff --git a/cmd/hauler/cli/cli.go b/cmd/hauler/cli/cli.go index 8c1d030..8efb89e 100644 --- a/cmd/hauler/cli/cli.go +++ b/cmd/hauler/cli/cli.go @@ -69,6 +69,7 @@ func New(ctx context.Context, ro *flags.CliRootOpts) *cobra.Command { cmd.AddCommand(cranecmd.NewCmdAuthLogin("hauler")) cmd.AddCommand(cranecmd.NewCmdAuthLogout("hauler")) addStore(cmd, ro) + addCopy(cmd, ro) addVersion(cmd, ro) addCompletion(cmd, ro) diff --git a/cmd/hauler/cli/copy.go b/cmd/hauler/cli/copy.go new file mode 100644 index 0000000..85c25ca --- /dev/null +++ b/cmd/hauler/cli/copy.go @@ -0,0 +1,176 @@ +package cli + +import ( + "context" + "fmt" + "time" + + "github.com/google/go-containerregistry/pkg/authn" + gname "github.com/google/go-containerregistry/pkg/name" + gv1 "github.com/google/go-containerregistry/pkg/v1" + "github.com/google/go-containerregistry/pkg/v1/remote" + "github.com/spf13/cobra" + + "hauler.dev/go/hauler/v2/internal/flags" + "hauler.dev/go/hauler/v2/pkg/audit" + "hauler.dev/go/hauler/v2/pkg/content" + "hauler.dev/go/hauler/v2/pkg/log" + "hauler.dev/go/hauler/v2/pkg/retry" + "hauler.dev/go/hauler/v2/pkg/store" +) + +func addCopy(parent *cobra.Command, ro *flags.CliRootOpts) { + o := &flags.ImageCopyOpts{} + + cmd := &cobra.Command{ + Use: "copy SRC DST", + Aliases: []string{"cp"}, + Short: "Copy an artifact between registries", + Example: ` # copy an image to another registry + hauler copy busybox:latest registry.example.com/busybox:latest + + # copy a specific platform out of a multi-arch image + hauler copy ghcr.io/hauler-dev/hauler-debug:v2.0.3 registry.example.com/hauler-debug:v2.0.3 --platform linux/amd64 + + # copy to a registry with a self-signed certificate + hauler copy busybox:latest registry.example.com/busybox:latest --insecure-skip-tls-verify + + # copy to a registry with no TLS at all + hauler copy busybox:latest registry.example.com/busybox:latest --plain-http`, + Args: cobra.ExactArgs(2), + RunE: func(cmd *cobra.Command, args []string) error { + return CopyImageCmd(cmd.Context(), o, args[0], args[1], ro) + }, + } + o.AddFlags(cmd) + parent.AddCommand(cmd) +} + +// CopyImageCmd copies src to dst directly, registry to registry, no store involved. +func CopyImageCmd(ctx context.Context, o *flags.ImageCopyOpts, src, dst string, ro *flags.CliRootOpts) error { + l := log.FromContext(ctx) + + retries, err := flags.ResolveRetries(o.Retries) + if err != nil { + return err + } + + tr, err := content.BuildTransport(o.InsecureSkipTLSVerify, o.CaFile) + if err != nil { + return err + } + + opts := []remote.Option{ + remote.WithAuthFromKeychain(authn.DefaultKeychain), + remote.WithContext(ctx), + remote.WithTransport(tr), + } + + var nameOpts []gname.Option + if o.PlainHTTP { + nameOpts = append(nameOpts, gname.Insecure) + } + + srcRef, err := gname.ParseReference(src, nameOpts...) + if err != nil { + return fmt.Errorf("parsing source reference %q: %w", src, err) + } + dstRef, err := gname.ParseReference(dst, nameOpts...) + if err != nil { + return fmt.Errorf("parsing destination reference %q: %w", dst, err) + } + + l.Infof("copying [%s] to [%s]", src, dst) + + start := time.Now() + var digest string + err = retry.Operation(ctx, &flags.StoreRootOpts{Retries: retries}, ro, func() error { + d, copyErr := copyOnce(srcRef, dstRef, o.Platform, opts) + if copyErr == nil { + digest = d + } + return copyErr + }) + if err != nil { + l.Errorf("unable to copy [%s] to [%s]: %v", src, dst, err) + return err + } + + if flags.AuditLevel(ro) != "none" { + e := audit.Entry{ + Command: "copy", + Args: []string{src, dst}, + Type: "image", + Reference: dst, + Digest: digest, + } + if flags.AuditLevel(ro) == "verbose" { + sys := audit.BuildSystem() + g := audit.BuildGlobal(ro, nil) + e.System = &sys + e.Global = &g + e.Flags = map[string]any{ + "insecure-skip-tls-verify": o.InsecureSkipTLSVerify, + "plain-http": o.PlainHTTP, + "ca-file": o.CaFile, + "platform": o.Platform, + } + } + if err := audit.Append(ro.HaulerDir, e); err != nil { + l.Warnf("failed to write audit entry: %v", err) + } + l.Debugf("generated audit id of [%s]", audit.ID()) + } else { + l.Debugf("generated audit id of [none]") + } + + l.Infof("✓ copied [%s] to [%s] (%.1fs)", src, dst, time.Since(start).Seconds()) + + return nil +} + +// copyOnce copies srcRef to dstRef and returns the digest copied. No +// platform filter keeps a multi-arch index intact; platform picks one child. +func copyOnce(srcRef, dstRef gname.Reference, platform string, opts []remote.Option) (string, error) { + desc, err := remote.Get(srcRef, opts...) + if err != nil { + return "", fmt.Errorf("fetching descriptor for %q: %w", srcRef.Name(), err) + } + + if idx, idxErr := desc.ImageIndex(); idxErr == nil && platform == "" { + if err := remote.WriteIndex(dstRef, idx, opts...); err != nil { + return "", fmt.Errorf("writing index for %q: %w", dstRef.Name(), err) + } + d, err := idx.Digest() + if err != nil { + return "", fmt.Errorf("getting index digest for %q: %w", srcRef.Name(), err) + } + return d.String(), nil + } + + var img gv1.Image + if platform != "" { + p, err := store.ParsePlatform(platform) + if err != nil { + return "", err + } + img, err = remote.Image(srcRef, append(append([]remote.Option{}, opts...), remote.WithPlatform(p))...) + if err != nil { + return "", fmt.Errorf("fetching image %q: %w", srcRef.Name(), err) + } + } else { + img, err = desc.Image() + if err != nil { + return "", fmt.Errorf("fetching image %q: %w", srcRef.Name(), err) + } + } + + if err := remote.Write(dstRef, img, opts...); err != nil { + return "", fmt.Errorf("writing image for %q: %w", dstRef.Name(), err) + } + d, err := img.Digest() + if err != nil { + return "", fmt.Errorf("getting image digest for %q: %w", srcRef.Name(), err) + } + return d.String(), nil +} diff --git a/cmd/hauler/cli/store.go b/cmd/hauler/cli/store.go index 4fd07e5..18f9392 100644 --- a/cmd/hauler/cli/store.go +++ b/cmd/hauler/cli/store.go @@ -310,8 +310,8 @@ func addStoreCopy(rso *flags.StoreRootOpts, ro *flags.CliRootOpts) *cobra.Comman Use: "copy", Short: "Copy all store content to another location", Example: ` # supported copy target prefixes - registry:// | reg:// | oci:// - Pushes the store to an OCI registry - directory:// | dir:// - Extracts the store to a directory`, + registry:// | reg:// | oci:// - Pushes the store to an OCI registry + directory:// | dir:// - Extracts the store to a directory`, Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { ctx := cmd.Context() @@ -354,13 +354,13 @@ func addStoreAddFile(rso *flags.StoreRootOpts, ro *flags.CliRootOpts) *cobra.Com Use: "file", Short: "Add a file to the store", Example: ` # fetch local file - hauler store add file file.txt + hauler store add file file.txt - # fetch remote file - hauler store add file https://get.rke2.io/install.sh + # fetch remote file + hauler store add file https://get.rke2.io/install.sh - # fetch remote file and assign new name - hauler store add file https://get.hauler.dev --name hauler-install.sh`, + # fetch remote file and assign new name + hauler store add file https://get.hauler.dev --name hauler-install.sh`, Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { ctx := cmd.Context() @@ -384,27 +384,27 @@ func addStoreAddImage(rso *flags.StoreRootOpts, ro *flags.CliRootOpts) *cobra.Co cmd := &cobra.Command{ Use: "image", Short: "Add a image to the store", - Example: ` # fetch image - hauler store add image busybox + Example: ` # fetch image + hauler store add image busybox - # fetch image with repository and tag - hauler store add image library/busybox:stable + # fetch image with repository and tag + hauler store add image library/busybox:stable - # fetch image with full image reference and specific platform - hauler store add image ghcr.io/hauler-dev/hauler-debug:v1.2.0 --platform linux/amd64 + # fetch image with full image reference and specific platform + hauler store add image ghcr.io/hauler-dev/hauler-debug:v1.2.0 --platform linux/amd64 - # fetch image with full image reference via digest - hauler store add image gcr.io/distroless/base@sha256:7fa7445dfbebae4f4b7ab0e6ef99276e96075ae42584af6286ba080750d6dfe5 + # fetch image with full image reference via digest + hauler store add image gcr.io/distroless/base@sha256:7fa7445dfbebae4f4b7ab0e6ef99276e96075ae42584af6286ba080750d6dfe5 - # fetch image with full image reference, specific platform, and signature verification - curl -sfOL https://raw.githubusercontent.com/rancherfederal/carbide-releases/main/carbide-key.pub - hauler store add image rgcrprod.azurecr.us/rancher/rke2-runtime:v1.31.5-rke2r1 --platform linux/amd64 --key carbide-key.pub + # fetch image with full image reference, specific platform, and signature verification + curl -sfOL https://raw.githubusercontent.com/rancherfederal/carbide-releases/main/carbide-key.pub + hauler store add image rgcrprod.azurecr.us/rancher/rke2-runtime:v1.31.5-rke2r1 --platform linux/amd64 --key carbide-key.pub - # fetch image and rewrite path - hauler store add image busybox --rewrite custom-path/busybox:latest + # fetch image and rewrite path + hauler store add image busybox --rewrite custom-path/busybox:latest - # add image from local Docker daemon - hauler store add image my-local-app:latest --local`, + # add image from local Docker daemon + hauler store add image my-local-app:latest --local`, Args: cobra.ExactArgs(1), PreRunE: func(cmd *cobra.Command, args []string) error { // Check for ca-file & insecure-skip-tls-verify env variables @@ -417,6 +417,12 @@ func addStoreAddImage(rso *flags.StoreRootOpts, ro *flags.CliRootOpts) *cobra.Co o.InsecureSkipTLSVerify = &b } } + + // resolve *bool: nil unless the user explicitly passed the flag + if cmd.Flags().Changed("insecure-skip-tls-verify") { + v, _ := cmd.Flags().GetBool("insecure-skip-tls-verify") + o.InsecureSkipTLSVerify = &v + } return nil }, RunE: func(cmd *cobra.Command, args []string) error { @@ -441,26 +447,26 @@ func addStoreAddChart(rso *flags.StoreRootOpts, ro *flags.CliRootOpts) *cobra.Co cmd := &cobra.Command{ Use: "chart", Short: "Add a helm chart to the store", - Example: ` # fetch local helm chart - hauler store add chart path/to/chart/directory --repo . + Example: ` # fetch local helm chart + hauler store add chart path/to/chart/directory --repo . - # fetch local compressed helm chart - hauler store add chart path/to/chart.tar.gz --repo . + # fetch local compressed helm chart + hauler store add chart path/to/chart.tar.gz --repo . - # fetch remote oci helm chart - hauler store add chart hauler-helm --repo oci://ghcr.io/hauler-dev + # fetch remote oci helm chart + hauler store add chart hauler-helm --repo oci://ghcr.io/hauler-dev - # fetch remote oci helm chart with version - hauler store add chart hauler-helm --repo oci://ghcr.io/hauler-dev --version 1.2.0 + # fetch remote oci helm chart with version + hauler store add chart hauler-helm --repo oci://ghcr.io/hauler-dev --version 1.2.0 - # fetch remote helm chart - hauler store add chart rancher --repo https://releases.rancher.com/server-charts/stable + # fetch remote helm chart + hauler store add chart rancher --repo https://releases.rancher.com/server-charts/stable - # fetch remote helm chart with specific version - hauler store add chart rancher --repo https://releases.rancher.com/server-charts/latest --version 2.10.1 + # fetch remote helm chart with specific version + hauler store add chart rancher --repo https://releases.rancher.com/server-charts/latest --version 2.10.1 - # fetch remote helm chart and rewrite path - hauler store add chart hauler-helm --repo oci://ghcr.io/hauler-dev --rewrite custom-path/hauler-chart:latest`, + # fetch remote helm chart and rewrite path + hauler store add chart hauler-helm --repo oci://ghcr.io/hauler-dev --rewrite custom-path/hauler-chart:latest`, Args: cobra.ExactArgs(1), PreRunE: func(cmd *cobra.Command, args []string) error { n, err := flags.ResolveConcurrency(cmd.Flags().Changed("concurrency"), o.Concurrency) @@ -502,25 +508,25 @@ func addStoreRemove(rso *flags.StoreRootOpts, ro *flags.CliRootOpts) *cobra.Comm Use: "remove ", Short: "Remove an artifact from the content store", Example: ` # remove an image using full store reference - hauler store info - hauler store remove index.docker.io/library/busybox:stable + hauler store info + hauler store remove index.docker.io/library/busybox:stable - # remove a chart using full store reference - hauler store info - hauler store remove hauler/rancher:2.8.4 + # remove a chart using full store reference + hauler store info + hauler store remove hauler/rancher:2.8.4 - # remove a file using full store reference - hauler store info - hauler store remove hauler/rke2-install.sh + # remove a file using full store reference + hauler store info + hauler store remove hauler/rke2-install.sh - # remove any artifact with the latest tag - hauler store remove :latest + # remove any artifact with the latest tag + hauler store remove :latest - # remove any artifact with 'busybox' in the reference - hauler store remove busybox + # remove any artifact with 'busybox' in the reference + hauler store remove busybox - # force remove without verification - hauler store remove busybox:latest --force`, + # force remove without verification + hauler store remove busybox:latest --force`, Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { ctx := cmd.Context() diff --git a/cmd/hauler/cli/store/add.go b/cmd/hauler/cli/store/add.go index 39adb20..f2bc14f 100644 --- a/cmd/hauler/cli/store/add.go +++ b/cmd/hauler/cli/store/add.go @@ -389,7 +389,7 @@ func storeLocalImage(ctx context.Context, s *store.Layout, i v1.Image, _ *flags. start := time.Now() ignoreErrors := flags.ShouldIgnoreErrors(ro) - l.Debugf("adding image [%s] from local Docker daemon to the store", i.Name) + l.Debugf("resolving image [%s] from local Docker daemon (rewrite=%q)", i.Name, rewrite) r, err := name.ParseReference(i.Name) if err != nil { @@ -480,7 +480,11 @@ func storeImage(ctx context.Context, s *store.Layout, i v1.Image, platform strin return err } - log.BaseFromContext(ctx).Debugf("adding image [%s] to the store", i.Name) + insecureSkipTLSVerify := derefInsecure(i.InsecureSkipTLSVerify) + caFile := i.CaFile + + log.BaseFromContext(ctx).Debugf("resolving image [%s] (platform=%q, excludeExtras=%t, verified=%t, insecureSkipTLSVerify=%t, caFile=%q, rewrite=%q, digest=%q)", + i.Name, platform, excludeExtras, verified, insecureSkipTLSVerify, caFile, rewrite, pinnedDigest) r, err := name.ParseReference(i.Name) if err != nil { @@ -493,9 +497,6 @@ func storeImage(ctx context.Context, s *store.Layout, i v1.Image, platform strin } } - insecureSkipTLSVerify := derefInsecure(i.InsecureSkipTLSVerify) - caFile := i.CaFile - // fetch image along with any associated signatures and attestations. // A fresh store.ImageStats is built inside the closure on every attempt, // not once outside it, so a failed attempt's partial layer/byte counts diff --git a/cmd/hauler/cli/store/audit.go b/cmd/hauler/cli/store/audit.go index c8195b2..2780bd4 100644 --- a/cmd/hauler/cli/store/audit.go +++ b/cmd/hauler/cli/store/audit.go @@ -4,11 +4,5 @@ import "hauler.dev/go/hauler/v2/internal/flags" // auditLevel returns the resolved audit level (none, standard, verbose) func auditLevel(ro *flags.CliRootOpts) string { - if ro == nil { - return "none" - } - if ro.AuditLevel == "" { - return "standard" - } - return ro.AuditLevel + return flags.AuditLevel(ro) } diff --git a/cmd/hauler/cli/store/sync_test.go b/cmd/hauler/cli/store/sync_test.go index b1f0b90..ded084c 100644 --- a/cmd/hauler/cli/store/sync_test.go +++ b/cmd/hauler/cli/store/sync_test.go @@ -2025,8 +2025,8 @@ func TestSyncImages_ErrorPropagation(t *testing.T) { // bug: with concurrency=1 and a bad ref placed before several good refs, the // bad job's failure cancels the errgroup's derived context. Jobs queued after // it never get a chance to call s.AddImage, so they must never log the -// "adding image [...] to the store" INFO line either -- otherwise a failed -// sync of 1 image looks like it attempted all of them. +// "resolving image [...]" line either -- otherwise a failed sync of 1 image +// looks like it attempted all of them. // // concurrency=1 makes cancellation deterministic: errgroup.SetLimit(1) means // each g.Go call blocks acquiring its semaphore slot until the previous job's @@ -2047,7 +2047,7 @@ func TestRunImageJobs_CancelledJobsDoNotLogAddingImage(t *testing.T) { s := newTestStore(t) var buf bytes.Buffer - // "adding image [...]" now logs at Debug (cmd/hauler/cli/store/add.go), + // "resolving image [...]" now logs at Debug (cmd/hauler/cli/store/add.go), // so this test needs Debug-level output visible. Per-logger .Level() is // not sufficient on its own: zerolog's Logger.should() gates on // max(logger.level, zerolog.GlobalLevel()) -- and GlobalLevel is @@ -2073,9 +2073,9 @@ func TestRunImageJobs_CancelledJobsDoNotLogAddingImage(t *testing.T) { t.Fatal("runImageJobs: expected error, got nil") } - got := strings.Count(buf.String(), "adding image [") + got := strings.Count(buf.String(), "resolving image [") if got != 1 { - t.Errorf("\"adding image [\" logged %d times, want exactly 1 (only the failed job should have attempted logging; the %d good jobs queued after it must never start)\nfull log:\n%s", got, nGood, buf.String()) + t.Errorf("\"resolving image [\" logged %d times, want exactly 1 (only the failed job should have attempted logging; the %d good jobs queued after it must never start)\nfull log:\n%s", got, nGood, buf.String()) } } diff --git a/internal/flags/audit_level.go b/internal/flags/audit_level.go new file mode 100644 index 0000000..d02d115 --- /dev/null +++ b/internal/flags/audit_level.go @@ -0,0 +1,12 @@ +package flags + +// AuditLevel returns the resolved audit level (none, standard, verbose). +func AuditLevel(ro *CliRootOpts) string { + if ro == nil { + return "none" + } + if ro.AuditLevel == "" { + return "standard" + } + return ro.AuditLevel +} diff --git a/internal/flags/imagecopy.go b/internal/flags/imagecopy.go new file mode 100644 index 0000000..6b8482f --- /dev/null +++ b/internal/flags/imagecopy.go @@ -0,0 +1,27 @@ +package flags + +import ( + "fmt" + + "github.com/spf13/cobra" + "hauler.dev/go/hauler/v2/pkg/consts" +) + +// ImageCopyOpts holds flags for `hauler copy` -- not to be confused with CopyOpts (`store copy`). +type ImageCopyOpts struct { + InsecureSkipTLSVerify bool + PlainHTTP bool + CaFile string + Retries int + Platform string +} + +func (o *ImageCopyOpts) AddFlags(cmd *cobra.Command) { + f := cmd.Flags() + + f.BoolVar(&o.InsecureSkipTLSVerify, "insecure-skip-tls-verify", false, "(Optional) Skip TLS certificate verification") + f.BoolVar(&o.PlainHTTP, "plain-http", false, "(Optional) Allow plain HTTP connections") + f.StringVar(&o.CaFile, "ca-file", "", "(Optional) Location of CA Bundle to enable certification verification") + f.IntVarP(&o.Retries, "retries", "r", 0, fmt.Sprintf("Set the number of retries for operations (0 uses HAULER_RETRIES, otherwise defaults to %d)", consts.DefaultRetries)) + f.StringVarP(&o.Platform, "platform", "p", "", "(Optional) Specify the platform of the image... i.e. linux/amd64 (defaults to all)") +} diff --git a/pkg/content/copy.go b/pkg/content/copy.go new file mode 100644 index 0000000..ebc8f07 --- /dev/null +++ b/pkg/content/copy.go @@ -0,0 +1,158 @@ +package content + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + + "github.com/containerd/containerd/v2/core/remotes" + "github.com/containerd/errdefs" + ocispec "github.com/opencontainers/image-spec/specs-go/v1" + "github.com/rs/zerolog" + + "hauler.dev/go/hauler/v2/pkg/consts" +) + +// CopyDescriptorGraph recursively copies desc and everything it references from fetcher to pusher. +func CopyDescriptorGraph(ctx context.Context, desc ocispec.Descriptor, fetcher remotes.Fetcher, pusher remotes.Pusher) (err error) { + switch desc.MediaType { + case ocispec.MediaTypeImageManifest, consts.DockerManifestSchema2: + rc, err := fetcher.Fetch(ctx, desc) + if err != nil { + return fmt.Errorf("failed to fetch manifest: %w", err) + } + defer func() { + if closeErr := rc.Close(); closeErr != nil && err == nil { + err = fmt.Errorf("failed to close manifest reader: %w", closeErr) + } + }() + + data, err := io.ReadAll(rc) + if err != nil { + return fmt.Errorf("failed to read manifest: %w", err) + } + + var manifest ocispec.Manifest + if err := json.Unmarshal(data, &manifest); err != nil { + return fmt.Errorf("failed to unmarshal manifest: %w", err) + } + + if err := copyDescriptor(ctx, manifest.Config, fetcher, pusher); err != nil { + return fmt.Errorf("failed to copy config: %w", err) + } + + for _, layer := range manifest.Layers { + if err := copyDescriptor(ctx, layer, fetcher, pusher); err != nil { + return fmt.Errorf("failed to copy layer: %w", err) + } + } + + // push the manifest itself using the already-fetched data to avoid double-fetching + if err := pushData(ctx, desc, data, pusher); err != nil { + return fmt.Errorf("failed to push manifest: %w", err) + } + + case ocispec.MediaTypeImageIndex, consts.DockerManifestListSchema2: + rc, err := fetcher.Fetch(ctx, desc) + if err != nil { + return fmt.Errorf("failed to fetch index: %w", err) + } + defer func() { + if closeErr := rc.Close(); closeErr != nil && err == nil { + err = fmt.Errorf("failed to close index reader: %w", closeErr) + } + }() + + data, err := io.ReadAll(rc) + if err != nil { + return fmt.Errorf("failed to read index: %w", err) + } + + var index ocispec.Index + if err := json.Unmarshal(data, &index); err != nil { + return fmt.Errorf("failed to unmarshal index: %w", err) + } + + for _, child := range index.Manifests { + if err := CopyDescriptorGraph(ctx, child, fetcher, pusher); err != nil { + return fmt.Errorf("failed to copy child: %w", err) + } + } + + // push the index itself using the already-fetched data to avoid double-fetching + if err := pushData(ctx, desc, data, pusher); err != nil { + return fmt.Errorf("failed to push index: %w", err) + } + + default: + if err := copyDescriptor(ctx, desc, fetcher, pusher); err != nil { + return fmt.Errorf("failed to copy descriptor: %w", err) + } + } + + return nil +} + +// copyDescriptor copies a single descriptor from fetcher to pusher. +func copyDescriptor(ctx context.Context, desc ocispec.Descriptor, fetcher remotes.Fetcher, pusher remotes.Pusher) (err error) { + rc, err := fetcher.Fetch(ctx, desc) + if err != nil { + return err + } + defer func() { + if closeErr := rc.Close(); closeErr != nil && err == nil { + err = fmt.Errorf("failed to close reader: %w", closeErr) + } + }() + + writer, err := pusher.Push(ctx, desc) + if err != nil { + if errdefs.IsAlreadyExists(err) { + zerolog.Ctx(ctx).Debug().Msgf("existing blob: %s", desc.Digest) + return nil // content already present on remote + } + return err + } + defer func() { + if closeErr := writer.Close(); closeErr != nil && err == nil { + err = closeErr + } + }() + + n, err := io.Copy(writer, rc) + if err != nil { + return err + } + + if err := writer.Commit(ctx, n, desc.Digest); err != nil { + return err + } + zerolog.Ctx(ctx).Debug().Msgf("pushed blob: %s", desc.Digest) + return nil +} + +// pushData pushes already-fetched data to the pusher without re-fetching -- +// used once a manifest/index's bytes are already in hand from parsing it. +func pushData(ctx context.Context, desc ocispec.Descriptor, data []byte, pusher remotes.Pusher) (err error) { + writer, err := pusher.Push(ctx, desc) + if err != nil { + if errdefs.IsAlreadyExists(err) { + return nil // content already present on remote + } + return fmt.Errorf("failed to get writer: %w", err) + } + defer func() { + if closeErr := writer.Close(); closeErr != nil && err == nil { + err = fmt.Errorf("failed to close writer: %w", closeErr) + } + }() + + n, err := io.Copy(writer, bytes.NewReader(data)) + if err != nil { + return fmt.Errorf("failed to write data: %w", err) + } + + return writer.Commit(ctx, n, desc.Digest) +} diff --git a/pkg/store/store.go b/pkg/store/store.go index 7804186..0038f8a 100644 --- a/pkg/store/store.go +++ b/pkg/store/store.go @@ -1,7 +1,6 @@ package store import ( - "bytes" "context" "encoding/json" "fmt" @@ -10,8 +9,6 @@ import ( "path/filepath" "strings" - "github.com/containerd/containerd/v2/core/remotes" - "github.com/containerd/errdefs" "github.com/google/go-containerregistry/pkg/authn" gname "github.com/google/go-containerregistry/pkg/name" v1 "github.com/google/go-containerregistry/pkg/v1" @@ -298,7 +295,7 @@ func (l *Layout) AddImage(ctx context.Context, ref string, platform string, excl // verified index, so the chain of trust holds. imgOpts := append([]remote.Option{}, allOpts...) if platform != "" { - p, err := parsePlatform(platform) + p, err := ParsePlatform(platform) if err != nil { return "", err } @@ -640,8 +637,8 @@ func (l *Layout) saveRelatedArtifacts(ctx context.Context, ref gname.Reference, return saved, nil } -// parsePlatform parses a platform string in "os/arch[/variant]" format into a v1.Platform. -func parsePlatform(s string) (v1.Platform, error) { +// ParsePlatform parses a platform string in "os/arch[/variant]" format into a v1.Platform. +func ParsePlatform(s string) (v1.Platform, error) { parts := strings.SplitN(s, "/", 3) if len(parts) < 2 { return v1.Platform{}, fmt.Errorf("invalid platform %q: expected os/arch[/variant]", s) @@ -698,168 +695,14 @@ func (l *Layout) Copy(ctx context.Context, ref string, to content.Target, toRef } // Recursively copy the descriptor graph (matches oras.Copy behavior) - if err := l.copyDescriptorGraph(ctx, desc, fetcher, pusher); err != nil { + if err := content.CopyDescriptorGraph(ctx, desc, fetcher, pusher); err != nil { return ocispec.Descriptor{}, err } return desc, nil } -// copyDescriptorGraph recursively copies a descriptor and all its referenced content -// This matches the behavior of oras.Copy by walking the entire descriptor graph -func (l *Layout) copyDescriptorGraph(ctx context.Context, desc ocispec.Descriptor, fetcher remotes.Fetcher, pusher remotes.Pusher) (err error) { - switch desc.MediaType { - case ocispec.MediaTypeImageManifest, consts.DockerManifestSchema2: - // Fetch and parse the manifest - rc, err := fetcher.Fetch(ctx, desc) - if err != nil { - return fmt.Errorf("failed to fetch manifest: %w", err) - } - defer func() { - if closeErr := rc.Close(); closeErr != nil && err == nil { - err = fmt.Errorf("failed to close manifest reader: %w", closeErr) - } - }() - - data, err := io.ReadAll(rc) - if err != nil { - return fmt.Errorf("failed to read manifest: %w", err) - } - - var manifest ocispec.Manifest - if err := json.Unmarshal(data, &manifest); err != nil { - return fmt.Errorf("failed to unmarshal manifest: %w", err) - } - - // Copy config blob - if err := l.copyDescriptor(ctx, manifest.Config, fetcher, pusher); err != nil { - return fmt.Errorf("failed to copy config: %w", err) - } - - // Copy all layer blobs - for _, layer := range manifest.Layers { - if err := l.copyDescriptor(ctx, layer, fetcher, pusher); err != nil { - return fmt.Errorf("failed to copy layer: %w", err) - } - } - - // Push the manifest itself using the already-fetched data to avoid double-fetching - if err := l.pushData(ctx, desc, data, pusher); err != nil { - return fmt.Errorf("failed to push manifest: %w", err) - } - - case ocispec.MediaTypeImageIndex, consts.DockerManifestListSchema2: - // Fetch and parse the index - rc, err := fetcher.Fetch(ctx, desc) - if err != nil { - return fmt.Errorf("failed to fetch index: %w", err) - } - defer func() { - if closeErr := rc.Close(); closeErr != nil && err == nil { - err = fmt.Errorf("failed to close index reader: %w", closeErr) - } - }() - - data, err := io.ReadAll(rc) - if err != nil { - return fmt.Errorf("failed to read index: %w", err) - } - - var index ocispec.Index - if err := json.Unmarshal(data, &index); err != nil { - return fmt.Errorf("failed to unmarshal index: %w", err) - } - - // Recursively copy each child (could be manifest or nested index) - for _, child := range index.Manifests { - if err := l.copyDescriptorGraph(ctx, child, fetcher, pusher); err != nil { - return fmt.Errorf("failed to copy child: %w", err) - } - } - - // Push the index itself using the already-fetched data to avoid double-fetching - if err := l.pushData(ctx, desc, data, pusher); err != nil { - return fmt.Errorf("failed to push index: %w", err) - } - - default: - // For other types (config blobs, layers, etc.), just copy the blob - if err := l.copyDescriptor(ctx, desc, fetcher, pusher); err != nil { - return fmt.Errorf("failed to copy descriptor: %w", err) - } - } - - return nil -} - -// copyDescriptor copies a single descriptor from source to target -func (l *Layout) copyDescriptor(ctx context.Context, desc ocispec.Descriptor, fetcher remotes.Fetcher, pusher remotes.Pusher) (err error) { - // Fetch the content - rc, err := fetcher.Fetch(ctx, desc) - if err != nil { - return err - } - defer func() { - if closeErr := rc.Close(); closeErr != nil && err == nil { - err = fmt.Errorf("failed to close reader: %w", closeErr) - } - }() - - // Get a writer from the pusher - writer, err := pusher.Push(ctx, desc) - if err != nil { - if errdefs.IsAlreadyExists(err) { - zerolog.Ctx(ctx).Debug().Msgf("existing blob: %s", desc.Digest) - return nil // content already present on remote - } - return err - } - defer func() { - if closeErr := writer.Close(); closeErr != nil && err == nil { - err = closeErr - } - }() - - // Copy the content - n, err := io.Copy(writer, rc) - if err != nil { - return err - } - - // Commit the written content with the expected digest - if err := writer.Commit(ctx, n, desc.Digest); err != nil { - return err - } - zerolog.Ctx(ctx).Debug().Msgf("pushed blob: %s", desc.Digest) - return nil -} - -// pushData pushes already-fetched data to the pusher without re-fetching. -// This is used when we've already read the data for parsing and want to avoid double-fetching. -func (l *Layout) pushData(ctx context.Context, desc ocispec.Descriptor, data []byte, pusher remotes.Pusher) (err error) { - // Get a writer from the pusher - writer, err := pusher.Push(ctx, desc) - if err != nil { - if errdefs.IsAlreadyExists(err) { - return nil // content already present on remote - } - return fmt.Errorf("failed to get writer: %w", err) - } - defer func() { - if closeErr := writer.Close(); closeErr != nil && err == nil { - err = fmt.Errorf("failed to close writer: %w", closeErr) - } - }() - - // Write the data using io.Copy to handle short writes properly - n, err := io.Copy(writer, bytes.NewReader(data)) - if err != nil { - return fmt.Errorf("failed to write data: %w", err) - } - - // Commit the written content with the expected digest - return writer.Commit(ctx, n, desc.Digest) -} +// copyDescriptorGraph, copyDescriptor, and pushData moved to pkg/content.CopyDescriptorGraph. // CopyAll performs bulk copy operations on the stores oci layout to a provided target func (l *Layout) CopyAll(ctx context.Context, to content.Target, toMapper func(string) (string, error)) ([]ocispec.Descriptor, error) {