fix(dispatcher): only download materials that are in the CAS (#145)

Signed-off-by: Miguel Martinez Trivino <miguel@chainloop.dev>

Author: migmartri

app/cli/api/attestation/v1/crafting_state.pb.go

@@ -41,6 +41,9 @@ message Attestation {
         google.protobuf.Timestamp added_at = 5;
         workflowcontract.v1.CraftingSchema.Material.MaterialType material_type = 6; 
 
+        // Whether the material has been uploaded to the CAS
+        bool uploaded_to_cas = 7;
+
         message KeyVal {
             string id = 1 [(validate.rules).string.min_len = 1];
             string value = 2 [(validate.rules).string.min_len = 1];

app/cli/api/attestation/v1/crafting_state.pb.validate.go

@@ -138,7 +138,7 @@ func (r *WorkflowRunRepo) List(ctx context.Context, orgID, workflowID uuid.UUID,
 
 	workflowRuns, err := wfRunsQuery.All(ctx)
 	if err != nil {
-		return
+		return nil, "", err
 	}
 
 	for i, wr := range workflowRuns {

app/cli/api/attestation/v1/crafting_state.proto

@@ -199,7 +199,7 @@ func (d *FanOutDispatcher) Run(ctx context.Context, envelope *dsse.Envelope, org
 		// Retrieve material content
 		content := []byte(material.Value)
 		// It's a downloadable so we retrieve and override the content variable
-		if material.Hash != nil {
+		if material.Hash != nil && material.UploadedToCAS {
 			digest := material.Hash.String()
 			d.log.Infow("msg", "downloading material", "workflowID", workflowID, "materialType", material.Type, "name", material.Name)
 			buf := bytes.NewBuffer(nil)

app/controlplane/internal/data/workflowrun.go

@@ -18,13 +18,11 @@ package materials
 import (
 	"context"
 	"fmt"
-	"time"
 
 	api "github.com/chainloop-dev/chainloop/app/cli/api/attestation/v1"
 	schemaapi "github.com/chainloop-dev/chainloop/app/controlplane/api/workflowcontract/v1"
 	"github.com/chainloop-dev/chainloop/internal/casclient"
 	"github.com/rs/zerolog"
-	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 type ArtifactCrafter struct {
@@ -43,21 +41,5 @@ func NewArtifactCrafter(schema *schemaapi.CraftingSchema_Material, uploader casc
 
 // Craft will calculate the digest of the artifact, simulate an upload and return the material definition
 func (i *ArtifactCrafter) Craft(ctx context.Context, artifactPath string) (*api.Attestation_Material, error) {
-	result, err := i.uploader.UploadFile(ctx, artifactPath)
-	if err != nil {
-		i.logger.Debug().Err(err)
-		return nil, err
-	}
-
-	res := &api.Attestation_Material{
-		AddedAt:      timestamppb.New(time.Now()),
-		MaterialType: i.input.Type,
-		M: &api.Attestation_Material_Artifact_{
-			Artifact: &api.Attestation_Material_Artifact{
-				Id: i.input.Name, Digest: result.Digest, Name: result.Filename, IsSubject: i.input.Output,
-			},
-		},
-	}
-
-	return res, nil
+	return uploadAndCraft(ctx, i.input, i.uploader, artifactPath)
 }

app/controlplane/internal/dispatcher/dispatcher.go

@@ -87,6 +87,7 @@ func TestArtifactCraft(t *testing.T) {
 	got, err := crafter.Craft(context.TODO(), "file.txt")
 	assert.NoError(err)
 	assert.Equal(contractAPI.CraftingSchema_Material_ARTIFACT.String(), got.MaterialType.String())
+	assert.True(got.UploadedToCas)
 	assert.WithinDuration(time.Now(), got.AddedAt.AsTime(), 5*time.Second)
 
 	// The result includes the digest reference

internal/attestation/crafter/materials/artifact.go

@@ -19,14 +19,12 @@ import (
 	"context"
 	"fmt"
 	"os"
-	"time"
 
 	cdx "github.com/CycloneDX/cyclonedx-go"
 	api "github.com/chainloop-dev/chainloop/app/cli/api/attestation/v1"
 	schemaapi "github.com/chainloop-dev/chainloop/app/controlplane/api/workflowcontract/v1"
 	"github.com/chainloop-dev/chainloop/internal/casclient"
 	"github.com/rs/zerolog"
-	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 type CyclonedxJSONCrafter struct {
@@ -64,21 +62,5 @@ func (i *CyclonedxJSONCrafter) Craft(ctx context.Context, filePath string) (*api
 		return nil, fmt.Errorf("invalid cyclonedx sbom file: %w", ErrInvalidMaterialType)
 	}
 
-	result, err := i.uploader.UploadFile(ctx, filePath)
-	if err != nil {
-		i.logger.Debug().Err(err)
-		return nil, err
-	}
-
-	res := &api.Attestation_Material{
-		AddedAt:      timestamppb.New(time.Now()),
-		MaterialType: i.input.Type,
-		M: &api.Attestation_Material_Artifact_{
-			Artifact: &api.Attestation_Material_Artifact{
-				Id: i.input.Name, Digest: result.Digest, Name: "sbom.cyclonedx.json",
-			},
-		},
-	}
-
-	return res, nil
+	return uploadAndCraft(ctx, i.input, i.uploader, filePath)
 }

internal/attestation/crafter/materials/artifact_test.go

@@ -121,8 +121,9 @@ func TestCyclonedxJSONCraft(t *testing.T) {
 			require.NoError(t, err)
 			assert.Equal(contractAPI.CraftingSchema_Material_SBOM_CYCLONEDX_JSON.String(), got.MaterialType.String())
 			assert.WithinDuration(time.Now(), got.AddedAt.AsTime(), 5*time.Second)
+			assert.True(got.UploadedToCas)
 
-			// // The result includes the digest reference
+			// The result includes the digest reference
 			assert.Equal(got.GetArtifact(), &attestationApi.Attestation_Material_Artifact{
 				Id: "test", Digest: "deadbeef", Name: "sbom.cyclonedx.json",
 			})

internal/attestation/crafter/materials/cyclonedxjson.go

@@ -13,6 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//nolint:dupl
 package materials_test
 
 import (
@@ -63,6 +64,7 @@ func TestNewJUnitXMLCrafter(t *testing.T) {
 		})
 	}
 }
+
 func TestJUnitXMLCraft(t *testing.T) {
 	testCases := []struct {
 		name     string
@@ -95,6 +97,7 @@ func TestJUnitXMLCraft(t *testing.T) {
 		Name: "test",
 		Type: contractAPI.CraftingSchema_Material_JUNIT_XML,
 	}
+
 	l := zerolog.Nop()
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
@@ -119,6 +122,7 @@ func TestJUnitXMLCraft(t *testing.T) {
 
 			require.NoError(t, err)
 			assert.Equal(contractAPI.CraftingSchema_Material_JUNIT_XML.String(), got.MaterialType.String())
+			assert.True(got.UploadedToCas)
 			assert.WithinDuration(time.Now(), got.AddedAt.AsTime(), 5*time.Second)
 
 			// The result includes the digest reference