refactor(sbom, vex): integrate new stages into build conveyor

.specify/feature.json

@@ -1,3 +1,3 @@
 {
-  "feature_directory": "specs/020-elf-signing-anchor-digest"
+  "feature_directory": "specs/020-sbom-vex-build-stages"
 }

AGENTS.md

@@ -80,8 +80,6 @@ Correct: `task test:unit paths="./pkg/sbom/..." -- -focus=MyTest`
 
 - `task deps:install:pm` — extract the pm binary (linux/amd64) to `dest=<path>` (default `./bin/pm`). Optional.
 
-`format` and `lint*` come from a remote taskfile ([werf/common-ci](https://github.com/werf/common-ci)), so they need `TASK_X_REMOTE_TASKFILES=1` and network access.
-
 ## Verifying changes (MANDATORY)
 
 After changing Go code, run these in order — `task format` mutates files, so it goes first:

pkg/build/artifact_propagation.go

@@ -0,0 +1,100 @@
+package build
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/werf/logboek"
+	"github.com/werf/werf/v2/pkg/image"
+	"github.com/werf/werf/v2/pkg/oci/artifact"
+	"github.com/werf/werf/v2/pkg/storage"
+	"github.com/werf/werf/v2/pkg/storage/manager"
+)
+
+func ensureAttachedArtifacts(ctx context.Context, repository, digest string) error {
+	if repository == "" || repository == storage.LocalStorageAddress || digest == "" {
+		return fmt.Errorf("artifact source descriptor is incomplete")
+	}
+
+	index, err := artifact.PullFallbackIndex(ctx, repository, digest)
+	if err != nil {
+		return fmt.Errorf("check source artifact index %s@%s: %w", repository, digest, err)
+	}
+
+	manifest, err := index.IndexManifest()
+	if err != nil {
+		return fmt.Errorf("read source artifact index %s@%s: %w", repository, digest, err)
+	}
+	if len(manifest.Manifests) == 0 {
+		return fmt.Errorf("source image %s@%s has no attached artifacts", repository, digest)
+	}
+
+	return nil
+}
+
+func propagateArtifacts(ctx context.Context, projectName, imageName string, source, destination *image.StageDesc, caches []storage.StagesStorage, sourceStorages ...storage.StagesStorage) error {
+	var sourceStorage storage.StagesStorage
+	if len(sourceStorages) > 0 {
+		sourceStorage = sourceStorages[0]
+	}
+	return propagateArtifactsWithManager(ctx, projectName, imageName, source, destination, caches, sourceStorage, nil, nil)
+}
+
+func propagateArtifactsWithManager(ctx context.Context, projectName, imageName string, source, destination *image.StageDesc, caches []storage.StagesStorage, sourceStorage, destinationStorage storage.StagesStorage, storageManager manager.StorageManagerInterface) error {
+	if source == nil || source.Info == nil {
+		return fmt.Errorf("source image descriptor is unavailable")
+	}
+	if source.Info.Repository == "" || source.Info.Repository == storage.LocalStorageAddress {
+		return nil
+	}
+
+	if sourceStorage == nil {
+		sourceStorage = &storage.RepoStagesStorage{RepoAddress: source.Info.Repository}
+	}
+
+	copyArtifacts := func(ctx context.Context, sourceStorage storage.StagesStorage, sourceDigest string, destinationStorage storage.StagesStorage, destinationDigest string) error {
+		if storageManager != nil {
+			return storageManager.CopyAttachedArtifacts(ctx, sourceStorage, sourceDigest, destinationStorage, destinationDigest)
+		}
+		return sourceStorage.CopyAttachedArtifacts(ctx, sourceStorage.Address(), sourceDigest, destinationStorage.Address(), destinationDigest)
+	}
+
+	if destination != nil && destination.Info != nil &&
+		destination.Info.Repository != "" &&
+		destination.Info.Repository != storage.LocalStorageAddress &&
+		destination.Info.Repository != source.Info.Repository {
+		if destinationStorage == nil || destinationStorage.Address() != destination.Info.Repository {
+			destinationStorage = &storage.RepoStagesStorage{RepoAddress: destination.Info.Repository}
+		}
+		if err := logboek.Context(ctx).Default().LogProcess("image %s: copy artifacts into final repo %s", imageName, destination.Info.Repository).DoError(func() error {
+			return copyArtifacts(ctx, sourceStorage, source.Info.GetDigest(), destinationStorage, destination.Info.GetDigest())
+		}); err != nil {
+			return fmt.Errorf("copy attached artifacts into final repo %s: %w", destination.Info.Repository, err)
+		}
+	}
+
+	for _, cache := range caches {
+		if cache == nil || cache.Address() == storage.LocalStorageAddress || cache.Address() == source.Info.Repository {
+			continue
+		}
+
+		destinationDigest := source.Info.GetDigest()
+		if projectName != "" && source.StageID != nil && source.StageID.Digest != "" {
+			cacheDesc, err := cache.GetStageDesc(ctx, projectName, *source.StageID)
+			if err != nil || cacheDesc == nil || cacheDesc.Info == nil {
+				if err == nil {
+					err = fmt.Errorf("cache stage descriptor is unavailable")
+				}
+				logboek.Context(ctx).Warn().LogF("Warning: unable to resolve destination descriptor in cache stages storage %s: %s\n", cache.String(), err)
+				continue
+			}
+			destinationDigest = cacheDesc.Info.GetDigest()
+		}
+
+		if err := copyArtifacts(ctx, sourceStorage, source.Info.GetDigest(), cache, destinationDigest); err != nil {
+			logboek.Context(ctx).Warn().LogF("Warning: unable to copy artifacts into cache stages storage %s: %s\n", cache.String(), err)
+		}
+	}
+
+	return nil
+}

pkg/build/artifact_propagation_test.go

@@ -0,0 +1,258 @@
+package build
+
+import (
+	"bytes"
+	"context"
+	"net/http/httptest"
+	"strings"
+
+	"github.com/google/go-containerregistry/pkg/authn"
+	"github.com/google/go-containerregistry/pkg/name"
+	"github.com/google/go-containerregistry/pkg/registry"
+	"github.com/google/go-containerregistry/pkg/v1/random"
+	"github.com/google/go-containerregistry/pkg/v1/remote"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"go.uber.org/mock/gomock"
+
+	"github.com/werf/logboek"
+	"github.com/werf/werf/v2/pkg/attestation"
+	"github.com/werf/werf/v2/pkg/docker_registry"
+	"github.com/werf/werf/v2/pkg/image"
+	"github.com/werf/werf/v2/pkg/oci/artifact"
+	"github.com/werf/werf/v2/pkg/storage"
+	"github.com/werf/werf/v2/pkg/storage/manager"
+	"github.com/werf/werf/v2/test/mock"
+)
+
+type recordingArtifactStorageManager struct {
+	manager.StorageManagerInterface
+	copyCalls             int
+	lastSourceDigest      string
+	lastDestinationDigest string
+}
+
+func (m *recordingArtifactStorageManager) CopyAttachedArtifacts(_ context.Context, _ storage.StagesStorage, sourceDigest string, _ storage.StagesStorage, destinationDigest string) error {
+	m.copyCalls++
+	m.lastSourceDigest = sourceDigest
+	m.lastDestinationDigest = destinationDigest
+	return nil
+}
+
+var _ = Describe("artifact propagation", func() {
+	It("rejects an incomplete artifact source descriptor", func(ctx SpecContext) {
+		err := ensureAttachedArtifacts(ctx, "", "")
+
+		Expect(err).To(MatchError("artifact source descriptor is incomplete"))
+	})
+
+	It("rejects a nil source descriptor", func(ctx SpecContext) {
+		err := propagateArtifacts(ctx, "project", "app", nil, nil, nil)
+
+		Expect(err).To(MatchError("source image descriptor is unavailable"))
+	})
+
+	It("routes production propagation through StorageManager", func(ctx SpecContext) {
+		storageManager := &recordingArtifactStorageManager{}
+		source := &image.StageDesc{Info: &image.Info{Repository: "registry.example/source", RepoDigest: "registry.example/source@sha256:source"}}
+		destination := &image.StageDesc{Info: &image.Info{Repository: "registry.example/final", RepoDigest: "registry.example/final@sha256:destination"}}
+
+		Expect(propagateArtifactsWithManager(ctx, "project", "app", source, destination, nil, nil, nil, storageManager)).To(Succeed())
+		Expect(storageManager.copyCalls).To(Equal(1))
+	})
+
+	It("propagates a multi-platform index using its resolved destination digest", func(ctx SpecContext) {
+		storageManager := &recordingArtifactStorageManager{}
+		source := &image.StageDesc{
+			StageID: image.NewStageID("stage-digest", 1),
+			Info: &image.Info{
+				IsIndex:    true,
+				Repository: "registry.example/source",
+				RepoDigest: "registry.example/source@sha256:source-index",
+			},
+		}
+		cache := mock.NewMockStagesStorage(gomock.NewController(GinkgoT()))
+		cache.EXPECT().Address().Return("registry.example/cache").AnyTimes()
+		cache.EXPECT().String().Return("registry.example/cache").AnyTimes()
+		cache.EXPECT().GetStageDesc(ctx, "project", *source.StageID).Return(&image.StageDesc{Info: &image.Info{
+			IsIndex:    true,
+			Repository: "registry.example/cache",
+			RepoDigest: "registry.example/cache@sha256:destination-index",
+		}}, nil)
+
+		Expect(propagateArtifactsWithManager(ctx, "project", "app", source, nil, []storage.StagesStorage{cache}, nil, nil, storageManager)).To(Succeed())
+		Expect(storageManager.copyCalls).To(Equal(1))
+		Expect(storageManager.lastSourceDigest).To(Equal("sha256:source-index"))
+		Expect(storageManager.lastDestinationDigest).To(Equal("sha256:destination-index"))
+	})
+
+	It("skips local-only artifact sources", func(ctx SpecContext) {
+		err := propagateArtifacts(ctx, "project", "app", &image.StageDesc{
+			Info: &image.Info{Repository: ":local", RepoDigest: ":local@sha256:local"},
+		}, nil, nil)
+
+		Expect(err).To(Succeed())
+	})
+
+	Describe("propagation errors", func() {
+		var (
+			server       *httptest.Server
+			sourceRepo   string
+			sourceDigest string
+			remoteOpts   []remote.Option
+		)
+
+		stageDescFor := func(repo, digest string) *image.StageDesc {
+			return &image.StageDesc{Info: &image.Info{
+				Repository: repo,
+				RepoDigest: repo + "@" + digest,
+			}}
+		}
+
+		pushImageToRepo := func(ctx SpecContext, repo string) string {
+			img, err := random.Image(256, 1)
+			Expect(err).To(Succeed())
+			ref, err := name.NewTag(repo + ":v1")
+			Expect(err).To(Succeed())
+			Expect(remote.Write(ref, img, append([]remote.Option{remote.WithContext(ctx)}, remoteOpts...)...)).To(Succeed())
+			digest, err := img.Digest()
+			Expect(err).To(Succeed())
+			return digest.String()
+		}
+
+		BeforeEach(func(ctx SpecContext) {
+			Expect(docker_registry.Init(ctx, false, false, nil, nil)).To(Succeed())
+
+			server = httptest.NewServer(registry.New())
+			host := strings.TrimPrefix(server.URL, "http://")
+			sourceRepo = host + "/test/source"
+			remoteOpts = []remote.Option{remote.WithAuth(authn.Anonymous)}
+
+			img, err := random.Image(256, 1)
+			Expect(err).To(Succeed())
+			ref, err := name.NewTag(sourceRepo + ":v1")
+			Expect(err).To(Succeed())
+			Expect(remote.Write(ref, img, append([]remote.Option{remote.WithContext(ctx)}, remoteOpts...)...)).To(Succeed())
+			digest, err := img.Digest()
+			Expect(err).To(Succeed())
+			sourceDigest = digest.String()
+
+			store := artifact.NewOCIStore(sourceRepo, "app", remoteOpts...)
+			Expect(store.Attach(ctx, sourceDigest, attestation.DSSEMediaType, []byte(`{"v":1}`), "checksum-v1", "", "")).To(Succeed())
+		})
+
+		AfterEach(func() {
+			server.Close()
+		})
+
+		It("returns a final propagation error", func(ctx SpecContext) {
+			err := propagateArtifacts(ctx, "project", "app", stageDescFor(sourceRepo, sourceDigest), stageDescFor("127.0.0.1:1/unreachable/final", sourceDigest), nil)
+
+			Expect(err).To(MatchError(ContainSubstring("copy attached artifacts into final repo")))
+		})
+
+		It("propagates artifacts to final and cache repositories", func(ctx SpecContext) {
+			finalRepo := strings.TrimPrefix(server.URL, "http://") + "/test/final"
+			cacheRepo := strings.TrimPrefix(server.URL, "http://") + "/test/cache"
+			finalDigest := pushImageToRepo(ctx, finalRepo)
+			cacheDigest := pushImageToRepo(ctx, cacheRepo)
+			cache := mock.NewMockStagesStorage(gomock.NewController(GinkgoT()))
+			cache.EXPECT().Address().Return(cacheRepo).AnyTimes()
+			cache.EXPECT().String().Return(cacheRepo).AnyTimes()
+			cache.EXPECT().GetStageDesc(gomock.Any(), "project", image.StageID{Digest: "stage-digest"}).Return(stageDescFor(cacheRepo, cacheDigest), nil)
+
+			source := stageDescFor(sourceRepo, sourceDigest)
+			source.StageID = &image.StageID{Digest: "stage-digest"}
+			err := propagateArtifacts(ctx, "project", "app", source, stageDescFor(finalRepo, finalDigest), []storage.StagesStorage{cache})
+			Expect(err).To(Succeed())
+
+			for _, destination := range []struct {
+				repo   string
+				digest string
+			}{
+				{repo: finalRepo, digest: finalDigest},
+				{repo: cacheRepo, digest: cacheDigest},
+			} {
+				store := artifact.NewOCIStore(destination.repo, "app", remoteOpts...)
+				content, err := store.GetAttachedContent(ctx, destination.digest, attestation.DSSEMediaType, nil)
+				Expect(err).To(Succeed())
+				Expect(content).To(MatchJSON(`{"v":1}`))
+			}
+		})
+
+		It("resolves the cache destination digest before propagation", func(ctx SpecContext) {
+			cacheRepo := strings.TrimPrefix(server.URL, "http://") + "/test/cache-digest"
+			cacheDigest := pushImageToRepo(ctx, cacheRepo)
+			cache := mock.NewMockStagesStorage(gomock.NewController(GinkgoT()))
+			cache.EXPECT().Address().Return(cacheRepo).AnyTimes()
+			cache.EXPECT().String().Return(cacheRepo).AnyTimes()
+			cache.EXPECT().GetStageDesc(gomock.Any(), "project", gomock.Any()).Return(stageDescFor(cacheRepo, cacheDigest), nil)
+
+			source := stageDescFor(sourceRepo, sourceDigest)
+			source.StageID = &image.StageID{Digest: "stage-digest"}
+			Expect(propagateArtifacts(ctx, "project", "app", source, nil, []storage.StagesStorage{cache})).To(Succeed())
+
+			store := artifact.NewOCIStore(cacheRepo, "app", remoteOpts...)
+			content, err := store.GetAttachedContent(ctx, cacheDigest, attestation.DSSEMediaType, nil)
+			Expect(err).To(Succeed())
+			Expect(content).To(MatchJSON(`{"v":1}`))
+		})
+
+		It("skips propagation when the destination repository is identical", func(ctx SpecContext) {
+			destination := stageDescFor(sourceRepo, "sha256:does-not-exist")
+			Expect(propagateArtifacts(ctx, "project", "app", stageDescFor(sourceRepo, sourceDigest), destination, nil)).To(Succeed())
+		})
+
+		It("deduplicates an artifact with the same identity", func(ctx SpecContext) {
+			destinationRepo := strings.TrimPrefix(server.URL, "http://") + "/test/dedup"
+			destinationDigest := pushImageToRepo(ctx, destinationRepo)
+			destinationStore := artifact.NewOCIStore(destinationRepo, "app", remoteOpts...)
+			Expect(destinationStore.Attach(ctx, destinationDigest, attestation.DSSEMediaType, []byte(`{"v":2}`), "checksum-v1", "", "")).To(Succeed())
+
+			Expect(propagateArtifacts(ctx, "project", "app", stageDescFor(sourceRepo, sourceDigest), stageDescFor(destinationRepo, destinationDigest), nil)).To(Succeed())
+
+			content, err := destinationStore.GetAttachedContent(ctx, destinationDigest, attestation.DSSEMediaType, nil)
+			Expect(err).To(Succeed())
+			Expect(content).To(MatchJSON(`{"v":2}`))
+			index, err := artifact.PullFallbackIndex(ctx, destinationRepo, destinationDigest, remoteOpts...)
+			Expect(err).To(Succeed())
+			manifest, err := index.IndexManifest()
+			Expect(err).To(Succeed())
+			Expect(manifest.Manifests).To(HaveLen(1))
+		})
+
+		It("restores artifacts from a secondary repository onto the primary digest", func(ctx SpecContext) {
+			primaryRepo := strings.TrimPrefix(server.URL, "http://") + "/test/primary"
+			primaryDigest := pushImageToRepo(ctx, primaryRepo)
+			source := stageDescFor(sourceRepo, sourceDigest)
+			destination := stageDescFor(primaryRepo, primaryDigest)
+
+			Expect(ensureAttachedArtifacts(ctx, source.Info.Repository, source.Info.GetDigest())).To(Succeed())
+			Expect(propagateArtifacts(ctx, "project", "app", source, destination, nil)).To(Succeed())
+
+			store := artifact.NewOCIStore(primaryRepo, "app", remoteOpts...)
+			content, err := store.GetAttachedContent(ctx, primaryDigest, attestation.DSSEMediaType, nil)
+			Expect(err).To(Succeed())
+			Expect(content).To(MatchJSON(`{"v":1}`))
+		})
+
+		It("rejects a secondary source image without attached artifacts", func(ctx SpecContext) {
+			repo := strings.TrimPrefix(server.URL, "http://") + "/test/missing-artifacts"
+			digest := pushImageToRepo(ctx, repo)
+
+			err := ensureAttachedArtifacts(ctx, repo, digest)
+			Expect(err).To(MatchError(ContainSubstring("has no attached artifacts")))
+		})
+
+		It("logs cache propagation errors and continues", func(ctx SpecContext) {
+			var output bytes.Buffer
+			logCtx := logboek.NewContext(ctx, logboek.NewLogger(&output, &output))
+			cache := mock.NewMockStagesStorage(gomock.NewController(GinkgoT()))
+			cache.EXPECT().Address().Return("127.0.0.1:1/unreachable/cache").AnyTimes()
+			cache.EXPECT().String().Return("127.0.0.1:1/unreachable/cache").AnyTimes()
+
+			Expect(propagateArtifacts(logCtx, "", "app", stageDescFor(sourceRepo, sourceDigest), nil, []storage.StagesStorage{cache})).To(Succeed())
+			Expect(output.String()).To(ContainSubstring("Warning: unable to copy artifacts into cache stages storage"))
+		})
+	})
+})

pkg/build/artifact_stage_lifecycle.go

@@ -0,0 +1,22 @@
+package build
+
+import (
+	"context"
+
+	"github.com/werf/werf/v2/pkg/build/image"
+	"github.com/werf/werf/v2/pkg/build/stage"
+)
+
+func runRestoredArtifactStages(ctx context.Context, img *image.Image, phase Phase) error {
+	for _, stg := range img.GetStages() {
+		if artifactStage, ok := stg.(interface {
+			GetArtifactMetadata() *stage.ArtifactStageMetadata
+		}); !ok || artifactStage.GetArtifactMetadata() == nil {
+			continue
+		}
+		if err := phase.OnImageStage(ctx, img, stg); err != nil {
+			return err
+		}
+	}
+	return nil
+}