diff --git a/pkg/extensions/sync/on_demand.go b/pkg/extensions/sync/on_demand.go index fb9fbb8a..dcf4585b 100644 --- a/pkg/extensions/sync/on_demand.go +++ b/pkg/extensions/sync/on_demand.go @@ -12,8 +12,6 @@ import ( "github.com/containers/image/v5/copy" "github.com/containers/image/v5/signature" "github.com/containers/image/v5/types" - "github.com/opencontainers/go-digest" - ispec "github.com/opencontainers/image-spec/specs-go/v1" "zotregistry.io/zot/pkg/common" syncconf "zotregistry.io/zot/pkg/extensions/config/sync" @@ -27,11 +25,13 @@ const ( ) type syncContextUtils struct { - policyCtx *signature.PolicyContext - localCtx *types.SystemContext - upstreamCtx *types.SystemContext - upstreamAddr string - copyOptions copy.Options + policyCtx *signature.PolicyContext + localCtx *types.SystemContext + upstreamCtx *types.SystemContext + upstreamAddr string + copyOptions copy.Options + retryOptions *retry.Options + enforceSignatures bool } //nolint:gochecknoglobals @@ -142,7 +142,7 @@ func syncOneImage(ctx context.Context, imageChannel chan error, upstreamRepo = getRepoSource(localRepo, regCfg.Content[contentID]) } - retryOptions := &retry.RetryOptions{} + retryOptions := &retry.Options{} if regCfg.MaxRetries != nil { retryOptions.MaxRetry = *regCfg.MaxRetries @@ -188,6 +188,7 @@ func syncOneImage(ctx context.Context, imageChannel chan error, /* demanded object is a signature or artifact at tis point we already have images synced, but not their signatures. */ if isCosignTag(reference) || artifactType != "" { + //nolint: contextcheck err = syncSignaturesArtifacts(sig, localRepo, upstreamRepo, reference, artifactType) if err != nil { continue @@ -198,15 +199,23 @@ func syncOneImage(ctx context.Context, imageChannel chan error, return } - syncContextUtils := syncContextUtils{ - policyCtx: policyCtx, - localCtx: localCtx, - upstreamCtx: upstreamCtx, - upstreamAddr: upstreamAddr, - copyOptions: options, + var enforeSignatures bool + if regCfg.OnlySigned != nil && *regCfg.OnlySigned { + enforeSignatures = true } + + syncContextUtils := syncContextUtils{ + policyCtx: policyCtx, + localCtx: localCtx, + upstreamCtx: upstreamCtx, + upstreamAddr: upstreamAddr, + copyOptions: options, + retryOptions: &retry.Options{}, // we don't want to retry inline + enforceSignatures: enforeSignatures, + } + //nolint:contextcheck - skipped, copyErr := syncRun(regCfg, localRepo, upstreamRepo, reference, syncContextUtils, sig, log) + skipped, copyErr := syncRun(localRepo, upstreamRepo, reference, syncContextUtils, sig, log) if skipped { continue } @@ -241,7 +250,7 @@ func syncOneImage(ctx context.Context, imageChannel chan error, time.Sleep(retryOptions.Delay) if err = retry.RetryIfNecessary(ctx, func() error { - _, err := syncRun(regCfg, localRepo, upstreamRepo, reference, syncContextUtils, sig, log) + _, err := syncRun(localRepo, upstreamRepo, reference, syncContextUtils, sig, log) return err }, retryOptions); err != nil { @@ -260,12 +269,9 @@ func syncOneImage(ctx context.Context, imageChannel chan error, imageChannel <- nil } -func syncRun(regCfg syncconf.RegistryConfig, - localRepo, upstreamRepo, reference string, utils syncContextUtils, sig *signaturesCopier, +func syncRun(localRepo, upstreamRepo, reference string, utils syncContextUtils, sig *signaturesCopier, log log.Logger, ) (bool, error) { - upstreamImageDigest, refIsDigest := parseReference(reference) - upstreamImageRef, err := getImageRef(utils.upstreamAddr, upstreamRepo, reference) if err != nil { log.Error().Str("errorType", common.TypeOf(err)). @@ -275,118 +281,19 @@ func syncRun(regCfg syncconf.RegistryConfig, return false, err } - manifestBuf, mediaType, err := getImageRefManifest(context.Background(), utils.upstreamCtx, upstreamImageRef, log) - if err != nil { - return false, err - } - - if !refIsDigest { - upstreamImageDigest = digest.FromBytes(manifestBuf) - } - - if !isSupportedMediaType(mediaType) { - if mediaType == ispec.MediaTypeArtifactManifest { - err = sig.syncOCIArtifact(localRepo, upstreamRepo, reference, manifestBuf) - if err != nil { - return false, err - } - } - - return false, nil - } - - // get upstream signatures - cosignManifest, err := sig.getCosignManifest(upstreamRepo, upstreamImageDigest.String()) - if err != nil { - log.Error().Str("errorType", common.TypeOf(err)). - Err(err).Msgf("couldn't get upstream image %s cosign manifest", upstreamImageRef.DockerReference()) - } - - index, err := sig.getOCIRefs(upstreamRepo, upstreamImageDigest.String()) - if err != nil { - log.Error().Str("errorType", common.TypeOf(err)). - Err(err).Msgf("couldn't get upstream image %s OCI references", upstreamImageRef.DockerReference()) - } - - // check if upstream image is signed - if cosignManifest == nil && len(getNotationManifestsFromOCIRefs(index)) == 0 { - // upstream image not signed - if regCfg.OnlySigned != nil && *regCfg.OnlySigned { - // skip unsigned images - log.Info().Msgf("skipping image without signature %s", upstreamImageRef.DockerReference()) - - return true, nil - } - } - imageStore := sig.storeController.GetImageStore(localRepo) localCachePath, err := getLocalCachePath(imageStore, localRepo) if err != nil { log.Error().Err(err).Msgf("couldn't get localCachePath for %s", localRepo) - } - - localImageRef, err := getLocalImageRef(localCachePath, localRepo, reference) - if err != nil { - log.Error().Str("errorType", common.TypeOf(err)). - Err(err).Msgf("couldn't obtain a valid image reference for reference %s/%s:%s", - localCachePath, localRepo, reference) return false, err } defer os.RemoveAll(localCachePath) - log.Info().Msgf("copying image %s to %s", upstreamImageRef.DockerReference(), localCachePath) - - _, err = copy.Image(context.Background(), utils.policyCtx, localImageRef, upstreamImageRef, &utils.copyOptions) - if err != nil { - log.Error().Str("errorType", common.TypeOf(err)). - Err(err).Msgf("error encountered while syncing on demand %s to %s", - upstreamImageRef.DockerReference(), localCachePath) - - return false, err - } - - err = pushSyncedLocalImage(localRepo, reference, localCachePath, imageStore, log) - if err != nil { - log.Error().Str("errorType", common.TypeOf(err)). - Err(err).Msgf("error while pushing synced cached image %s", - fmt.Sprintf("%s/%s:%s", localCachePath, localRepo, reference)) - - return false, err - } - - err = sig.syncOCIRefs(localRepo, upstreamRepo, upstreamImageDigest.String(), index) - if err != nil { - return false, err - } - - err = sig.syncCosignSignature(localRepo, upstreamRepo, upstreamImageDigest.String(), cosignManifest) - if err != nil { - log.Error().Str("errorType", common.TypeOf(err)). - Err(err).Msgf("couldn't copy image cosign signature %s/%s:%s", utils.upstreamAddr, upstreamRepo, reference) - - return false, err - } - - refs, err := sig.getORASRefs(upstreamRepo, upstreamImageDigest.String()) - if err != nil { - log.Error().Str("errorType", common.TypeOf(err)). - Err(err).Msgf("couldn't get upstream image %s ORAS references", upstreamImageRef.DockerReference()) - } - - err = sig.syncORASRefs(localRepo, upstreamRepo, upstreamImageDigest.String(), refs) - if err != nil { - log.Error().Str("errorType", common.TypeOf(err)). - Err(err).Msgf("couldn't copy image ORAS references %s/%s:%s", utils.upstreamAddr, upstreamRepo, reference) - - return false, err - } - - log.Info().Msgf("successfully synced %s/%s:%s", utils.upstreamAddr, upstreamRepo, reference) - - return false, nil + return syncImageWithRefs(context.Background(), localRepo, upstreamRepo, reference, upstreamImageRef, + utils, sig, localCachePath, log) } func syncSignaturesArtifacts(sig *signaturesCopier, localRepo, upstreamRepo, reference, artifactType string) error { diff --git a/pkg/extensions/sync/sync.go b/pkg/extensions/sync/sync.go index 3bfb7a36..e5be3647 100644 --- a/pkg/extensions/sync/sync.go +++ b/pkg/extensions/sync/sync.go @@ -2,7 +2,6 @@ package sync import ( "context" - "errors" "fmt" "io" "net/http" @@ -17,7 +16,6 @@ import ( "github.com/containers/image/v5/docker/reference" "github.com/containers/image/v5/signature" "github.com/containers/image/v5/types" - "github.com/opencontainers/go-digest" ispec "github.com/opencontainers/image-spec/specs-go/v1" zerr "zotregistry.io/zot/errors" @@ -269,125 +267,29 @@ func syncRegistry(ctx context.Context, regCfg syncconf.RegistryConfig, defer os.RemoveAll(localCachePath) for _, upstreamImageRef := range repoReference.imageReferences { - manifestBuf, mediaType, err := getImageRefManifest(ctx, upstreamCtx, upstreamImageRef, log) - if err != nil { - return err + var enforeSignatures bool + if regCfg.OnlySigned != nil && *regCfg.OnlySigned { + enforeSignatures = true } - upstreamImageDigest := digest.FromBytes(manifestBuf) + syncContextUtils := syncContextUtils{ + policyCtx: policyCtx, + localCtx: localCtx, + upstreamCtx: upstreamCtx, + upstreamAddr: upstreamAddr, + copyOptions: options, + retryOptions: &retry.Options{}, // we don't want to retry inline + enforceSignatures: enforeSignatures, + } tag := getTagFromRef(upstreamImageRef, log).Tag() - if !isSupportedMediaType(mediaType) { - if mediaType == ispec.MediaTypeArtifactManifest { - err = sig.syncOCIArtifact(localRepo, upstreamRepo, tag, manifestBuf) //nolint - if err != nil { - return err - } - } - + skipped, err := syncImageWithRefs(ctx, localRepo, upstreamRepo, tag, upstreamImageRef, + syncContextUtils, sig, localCachePath, log) + if skipped || err != nil { + // skip continue } - - // get upstream signatures - cosignManifest, err := sig.getCosignManifest(upstreamRepo, upstreamImageDigest.String()) - if err != nil && !errors.Is(err, zerr.ErrSyncReferrerNotFound) { - log.Error().Err(err).Msgf("couldn't get upstream image %s cosign manifest", upstreamImageRef.DockerReference()) - - return err - } - - index, err := sig.getOCIRefs(upstreamRepo, upstreamImageDigest.String()) - if err != nil && !errors.Is(err, zerr.ErrSyncReferrerNotFound) { - log.Error().Err(err).Msgf("couldn't get upstream image %s OCI references", upstreamImageRef.DockerReference()) - - return err - } - - // check if upstream image is signed - if cosignManifest == nil && len(getNotationManifestsFromOCIRefs(index)) == 0 { - // upstream image not signed - if regCfg.OnlySigned != nil && *regCfg.OnlySigned { - // skip unsigned images - log.Info().Msgf("skipping image without signature %s", upstreamImageRef.DockerReference()) - - continue - } - } - - skipImage, err := canSkipImage(localRepo, tag, upstreamImageDigest, imageStore, log) - if err != nil { - log.Error().Err(err).Msgf("couldn't check if the upstream image %s can be skipped", - upstreamImageRef.DockerReference()) - - return err - } - - if !skipImage { - // sync image - localImageRef, err := getLocalImageRef(localCachePath, localRepo, tag) - if err != nil { - log.Error().Str("errorType", common.TypeOf(err)). - Err(err).Msgf("couldn't obtain a valid image reference for reference %s/%s:%s", - localCachePath, localRepo, tag) - - return err - } - - log.Info().Msgf("copying image %s to %s", upstreamImageRef.DockerReference(), localCachePath) - - if err = retry.RetryIfNecessary(ctx, func() error { - _, err = copy.Image(ctx, policyCtx, localImageRef, upstreamImageRef, &options) - - return err - }, retryOptions); err != nil { - log.Error().Str("errorType", common.TypeOf(err)). - Err(err).Msgf("error while copying image %s to %s", - upstreamImageRef.DockerReference(), localCachePath) - - return err - } - - // push from cache to repo - err = pushSyncedLocalImage(localRepo, tag, localCachePath, imageStore, log) - if err != nil { - log.Error().Str("errorType", common.TypeOf(err)). - Err(err).Msgf("error while pushing synced cached image %s", - fmt.Sprintf("%s/%s:%s", localCachePath, localRepo, tag)) - - return err - } - } else { - log.Info().Msgf("already synced image %s, checking its signatures", upstreamImageRef.DockerReference()) - } - - // sync signatures - if err = retry.RetryIfNecessary(ctx, func() error { - err = sig.syncOCIRefs(localRepo, upstreamRepo, upstreamImageDigest.String(), index) - if err != nil { - return err - } - - refs, err := sig.getORASRefs(upstreamRepo, upstreamImageDigest.String()) - if err != nil && !errors.Is(err, zerr.ErrSyncReferrerNotFound) { - return err - } - - err = sig.syncORASRefs(localRepo, upstreamRepo, upstreamImageDigest.String(), refs) - if err != nil { - return err - } - - err = sig.syncCosignSignature(localRepo, upstreamRepo, upstreamImageDigest.String(), cosignManifest) - if err != nil { - return err - } - - return nil - }, retryOptions); err != nil { - log.Error().Str("errorType", common.TypeOf(err)). - Err(err).Msgf("couldn't copy referrer for %s", upstreamImageRef.DockerReference()) - } } } diff --git a/pkg/extensions/sync/sync_test.go b/pkg/extensions/sync/sync_test.go index 5c8a3b4f..9267f3a7 100644 --- a/pkg/extensions/sync/sync_test.go +++ b/pkg/extensions/sync/sync_test.go @@ -447,10 +447,22 @@ func TestORAS(t *testing.T) { err = os.Chmod(path.Join(destDir, testImage, "index.json"), 0o755) So(err, ShouldBeNil) - resp, err = resty.R().Get(getORASReferrersURL) + // trigger getORASRefs err + err = os.Chmod(path.Join(srcDir, testImage, "blobs/sha256", adigest.Encoded()), 0o000) + So(err, ShouldBeNil) + resp, err = resty.R().Get(destBaseURL + "/v2/" + testImage + "/manifests/" + digest.String()) So(err, ShouldBeNil) So(resp, ShouldNotBeEmpty) + So(resp.StatusCode(), ShouldEqual, http.StatusOK) + + err = os.Chmod(path.Join(srcDir, testImage, "blobs/sha256", adigest.Encoded()), 0o755) + So(err, ShouldBeNil) + + resp, err = resty.R().Get(getORASReferrersURL) + So(err, ShouldBeNil) + So(resp, ShouldNotBeEmpty) + So(resp.StatusCode(), ShouldEqual, http.StatusOK) var refs ReferenceList @@ -540,7 +552,7 @@ func TestOnDemand(t *testing.T) { So(err, ShouldBeNil) So(resp.StatusCode(), ShouldEqual, 404) - err = os.MkdirAll(path.Join(destDir, testImage), 0o000) + err = os.Chmod(path.Join(destDir, testImage), 0o000) if err != nil { panic(err) } @@ -558,7 +570,7 @@ func TestOnDemand(t *testing.T) { So(err, ShouldBeNil) So(resp.StatusCode(), ShouldEqual, 404) - err = os.MkdirAll(path.Join(destDir, testImage, sync.SyncBlobUploadDir), 0o000) + err = os.Chmod(path.Join(destDir, testImage, sync.SyncBlobUploadDir), 0o000) if err != nil { panic(err) } @@ -604,6 +616,16 @@ func TestOnDemand(t *testing.T) { } So(destTagsList, ShouldResemble, srcTagsList) + + // trigger canSkipImage error + err = os.Chmod(path.Join(destDir, testImage, "index.json"), 0o000) + if err != nil { + panic(err) + } + + resp, err = destClient.R().Get(destBaseURL + "/v2/" + testImage + "/manifests/" + testImageTag) + So(err, ShouldBeNil) + So(resp.StatusCode(), ShouldEqual, 500) }) } @@ -4237,6 +4259,8 @@ func TestSyncSignaturesDiff(t *testing.T) { time.Sleep(500 * time.Millisecond) } + time.Sleep(3 * time.Second) + splittedURL = strings.SplitAfter(destBaseURL, ":") destPort := splittedURL[len(splittedURL)-1] diff --git a/pkg/extensions/sync/utils.go b/pkg/extensions/sync/utils.go index ecf32584..243d6624 100644 --- a/pkg/extensions/sync/utils.go +++ b/pkg/extensions/sync/utils.go @@ -13,6 +13,8 @@ import ( "github.com/Masterminds/semver" glob "github.com/bmatcuk/doublestar/v4" + "github.com/containers/common/pkg/retry" + "github.com/containers/image/v5/copy" "github.com/containers/image/v5/docker" "github.com/containers/image/v5/docker/reference" "github.com/containers/image/v5/manifest" @@ -636,3 +638,129 @@ func getImageRefManifest(ctx context.Context, upstreamCtx *types.SystemContext, return manifestBuf, mediaType, nil } + +func syncImageWithRefs(ctx context.Context, localRepo, upstreamRepo, reference string, + upstreamImageRef types.ImageReference, utils syncContextUtils, sig *signaturesCopier, + localCachePath string, log log.Logger, +) (bool, error) { + var skipped bool + + imageStore := sig.storeController.GetImageStore(localRepo) + + manifestBuf, mediaType, err := getImageRefManifest(ctx, utils.upstreamCtx, upstreamImageRef, log) + if err != nil { + return skipped, err + } + + upstreamImageDigest := godigest.FromBytes(manifestBuf) + + if !isSupportedMediaType(mediaType) { + if mediaType == ispec.MediaTypeArtifactManifest { + err = sig.syncOCIArtifact(localRepo, upstreamRepo, reference, manifestBuf) //nolint + if err != nil { + return skipped, err + } + } + + return skipped, nil + } + + // get upstream signatures + cosignManifest, err := sig.getCosignManifest(upstreamRepo, upstreamImageDigest.String()) + if err != nil { + log.Error().Err(err).Msgf("couldn't get upstream image %s cosign manifest", upstreamImageRef.DockerReference()) + } + + index, err := sig.getOCIRefs(upstreamRepo, upstreamImageDigest.String()) + if err != nil { + log.Error().Err(err).Msgf("couldn't get upstream image %s OCI references", upstreamImageRef.DockerReference()) + } + + // check if upstream image is signed + if cosignManifest == nil && len(getNotationManifestsFromOCIRefs(index)) == 0 { + // upstream image not signed + if utils.enforceSignatures { + // skip unsigned images + log.Info().Msgf("skipping image without signature %s", upstreamImageRef.DockerReference()) + skipped = true + + return skipped, nil + } + } + + skipImage, err := canSkipImage(localRepo, upstreamImageDigest.String(), upstreamImageDigest, imageStore, log) + if err != nil { + log.Error().Err(err).Msgf("couldn't check if the upstream image %s can be skipped", + upstreamImageRef.DockerReference()) + } + + if !skipImage { + // sync image + localImageRef, err := getLocalImageRef(localCachePath, localRepo, reference) + if err != nil { + log.Error().Str("errorType", common.TypeOf(err)). + Err(err).Msgf("couldn't obtain a valid image reference for reference %s/%s:%s", + localCachePath, localRepo, reference) + + return skipped, err + } + + log.Info().Msgf("copying image %s to %s", upstreamImageRef.DockerReference(), localCachePath) + + if err = retry.RetryIfNecessary(ctx, func() error { + _, err = copy.Image(ctx, utils.policyCtx, localImageRef, upstreamImageRef, &utils.copyOptions) + + return err + }, utils.retryOptions); err != nil { + log.Error().Str("errorType", common.TypeOf(err)). + Err(err).Msgf("error while copying image %s to %s", + upstreamImageRef.DockerReference(), localCachePath) + + return skipped, err + } + + // push from cache to repo + err = pushSyncedLocalImage(localRepo, reference, localCachePath, imageStore, log) + if err != nil { + log.Error().Str("errorType", common.TypeOf(err)). + Err(err).Msgf("error while pushing synced cached image %s", + fmt.Sprintf("%s/%s:%s", localCachePath, localRepo, reference)) + + return skipped, err + } + } else { + log.Info().Msgf("already synced image %s, checking its signatures", upstreamImageRef.DockerReference()) + } + + // sync signatures + if err = retry.RetryIfNecessary(ctx, func() error { + err = sig.syncOCIRefs(localRepo, upstreamRepo, upstreamImageDigest.String(), index) + if err != nil { + return err + } + + refs, err := sig.getORASRefs(upstreamRepo, upstreamImageDigest.String()) + if err != nil && !errors.Is(err, zerr.ErrSyncReferrerNotFound) { + return err + } + + err = sig.syncORASRefs(localRepo, upstreamRepo, upstreamImageDigest.String(), refs) + if err != nil { + return err + } + + err = sig.syncCosignSignature(localRepo, upstreamRepo, upstreamImageDigest.String(), cosignManifest) + if err != nil { + return err + } + + return nil + }, utils.retryOptions); err != nil { + log.Error().Str("errorType", common.TypeOf(err)). + Err(err).Msgf("couldn't copy referrer for %s", upstreamImageRef.DockerReference()) + + return skipped, err + } + + return skipped, nil +}