Refactor some of our pipe-closing logic between all the types. (#552)

This got copy-pasta-ed a bit as we added a lot of new types. I refactored
this out so we have the logic only once.

Signed-off-by: Dan Lorenc <lorenc.d@gmail.com>

pkg/types/alpine/v0.0.1/entry.go

@@ -138,19 +138,7 @@ func (v *V001Entry) fetchExternalEntities(ctx context.Context) error {
 	defer hashR.Close()
 	defer apkR.Close()
 
-	closePipesOnError := func(err error) error {
-		pipeReaders := []*io.PipeReader{hashR, apkR}
-		pipeWriters := []*io.PipeWriter{hashW, apkW}
-		for idx := range pipeReaders {
-			if e := pipeReaders[idx].CloseWithError(err); e != nil {
-				log.Logger.Error(fmt.Errorf("error closing pipe: %w", e))
-			}
-			if e := pipeWriters[idx].CloseWithError(err); e != nil {
-				log.Logger.Error(fmt.Errorf("error closing pipe: %w", e))
-			}
-		}
-		return err
-	}
+	closePipesOnError := types.PipeCloser(hashR, hashW, apkR, apkW)
 
 	oldSHA := ""
 	if v.AlpineModel.Package.Hash != nil && v.AlpineModel.Package.Hash.Value != nil {

pkg/types/helm/v0.0.1/entry.go

@@ -142,22 +142,9 @@ func (v *V001Entry) fetchExternalEntities(ctx context.Context) error {
 	g, ctx := errgroup.WithContext(ctx)
 
 	provenanceR, provenanceW := io.Pipe()
-
 	defer provenanceR.Close()
 
-	closePipesOnError := func(err error) error {
-		pipeReaders := []*io.PipeReader{provenanceR}
-		pipeWriters := []*io.PipeWriter{provenanceW}
-		for idx := range pipeReaders {
-			if e := pipeReaders[idx].CloseWithError(err); e != nil {
-				log.Logger.Error(fmt.Errorf("error closing pipe: %w", e))
-			}
-			if e := pipeWriters[idx].CloseWithError(err); e != nil {
-				log.Logger.Error(fmt.Errorf("error closing pipe: %w", e))
-			}
-		}
-		return err
-	}
+	closePipesOnError := types.PipeCloser(provenanceR, provenanceW)
 
 	g.Go(func() error {
 		defer provenanceW.Close()

pkg/types/rekord/v0.0.1/entry.go

@@ -129,19 +129,7 @@ func (v *V001Entry) FetchExternalEntities(ctx context.Context) error {
 	defer hashR.Close()
 	defer sigR.Close()
 
-	closePipesOnError := func(err error) error {
-		pipeReaders := []*io.PipeReader{hashR, sigR}
-		pipeWriters := []*io.PipeWriter{hashW, sigW}
-		for idx := range pipeReaders {
-			if e := pipeReaders[idx].CloseWithError(err); e != nil {
-				log.Logger.Error(fmt.Errorf("error closing pipe: %w", e))
-			}
-			if e := pipeWriters[idx].CloseWithError(err); e != nil {
-				log.Logger.Error(fmt.Errorf("error closing pipe: %w", e))
-			}
-		}
-		return err
-	}
+	closePipesOnError := types.PipeCloser(hashR, hashW, sigR, sigW)
 
 	oldSHA := ""
 	if v.RekordObj.Data.Hash != nil && v.RekordObj.Data.Hash.Value != nil {
@@ -282,12 +270,6 @@ func (v *V001Entry) Canonicalize(ctx context.Context) ([]byte, error) {
 	if err := v.FetchExternalEntities(ctx); err != nil {
 		return nil, err
 	}
-	if v.sigObj == nil {
-		return nil, errors.New("signature object not initialized before canonicalization")
-	}
-	if v.keyObj == nil {
-		return nil, errors.New("key object not initialized before canonicalization")
-	}
 
 	canonicalEntry := models.RekordV001Schema{}
 
@@ -329,7 +311,7 @@ func (v *V001Entry) Canonicalize(ctx context.Context) ([]byte, error) {
 // validate performs cross-field validation for fields in object
 func (v V001Entry) validate() error {
 	sig := v.RekordObj.Signature
-	if sig == nil {
+	if v.RekordObj.Signature == nil {
 		return errors.New("missing signature")
 	}
 	if len(sig.Content) == 0 && sig.URL.String() == "" {

pkg/types/rpm/v0.0.1/entry.go

@@ -139,19 +139,7 @@ func (v *V001Entry) FetchExternalEntities(ctx context.Context) error {
 	defer sigR.Close()
 	defer rpmR.Close()
 
-	closePipesOnError := func(err error) error {
-		pipeReaders := []*io.PipeReader{hashR, sigR, rpmR}
-		pipeWriters := []*io.PipeWriter{hashW, sigW, rpmW}
-		for idx := range pipeReaders {
-			if e := pipeReaders[idx].CloseWithError(err); e != nil {
-				log.Logger.Error(fmt.Errorf("error closing pipe: %w", e))
-			}
-			if e := pipeWriters[idx].CloseWithError(err); e != nil {
-				log.Logger.Error(fmt.Errorf("error closing pipe: %w", e))
-			}
-		}
-		return err
-	}
+	closePipesOnError := types.PipeCloser(hashR, hashW, sigR, sigW, rpmR, rpmW)
 
 	oldSHA := ""
 	if v.RPMModel.Package.Hash != nil && v.RPMModel.Package.Hash.Value != nil {

pkg/types/tuf/v0.0.1/entry.go

@@ -161,19 +161,7 @@ func (v *V001Entry) fetchExternalEntities(ctx context.Context) error {
 	defer metaR.Close()
 	defer rootR.Close()
 
-	closePipesOnError := func(err error) error {
-		pipeReaders := []*io.PipeReader{metaR, rootR}
-		pipeWriters := []*io.PipeWriter{metaW, rootW}
-		for idx := range pipeReaders {
-			if e := pipeReaders[idx].CloseWithError(err); e != nil {
-				log.Logger.Error(fmt.Errorf("error closing pipe: %w", e))
-			}
-			if e := pipeWriters[idx].CloseWithError(err); e != nil {
-				log.Logger.Error(fmt.Errorf("error closing pipe: %w", e))
-			}
-		}
-		return err
-	}
+	closePipesOnError := types.PipeCloser(metaR, metaW, rootR, rootW)
 
 	// verify artifact signature
 	artifactFactory, err := pki.NewArtifactFactory(pki.Tuf)

pkg/types/types.go

@@ -22,6 +22,7 @@ import (
 	"sync"
 
 	"github.com/sigstore/rekor/pkg/generated/models"
+	"github.com/sigstore/rekor/pkg/log"
 )
 
 // TypeMap stores mapping between type strings and entry constructors
@@ -78,3 +79,18 @@ func ListImplementedTypes() []string {
 	})
 	return retVal
 }
+
+type errCloser interface {
+	CloseWithError(error) error
+}
+
+func PipeCloser(errClosers ...errCloser) func(err error) error {
+	return func(err error) error {
+		for _, p := range errClosers {
+			if err := p.CloseWithError(err); err != nil {
+				log.Logger.Error(fmt.Errorf("error closing pipe: %w", err))
+			}
+		}
+		return err
+	}
+}