Respond to authoratative CPEs from catalogers

internal/task/package_task_factory.go

@@ -15,10 +15,11 @@ import (
 	"github.com/anchore/syft/syft/artifact"
 	"github.com/anchore/syft/syft/cataloging"
 	"github.com/anchore/syft/syft/cataloging/pkgcataloging"
+	"github.com/anchore/syft/syft/cpe"
 	"github.com/anchore/syft/syft/event/monitor"
 	"github.com/anchore/syft/syft/file"
 	"github.com/anchore/syft/syft/pkg"
-	"github.com/anchore/syft/syft/pkg/cataloger/common/cpe"
+	cpeutils "github.com/anchore/syft/syft/pkg/cataloger/common/cpe"
 )
 
 type packageTaskFactory func(cfg CatalogingFactoryConfig) Task
@@ -109,15 +110,16 @@ func NewPackageTask(cfg CatalogingFactoryConfig, c pkg.Cataloger, tags ...string
 			if p.FoundBy == "" {
 				p.FoundBy = catalogerName
 			}
-			if cfg.DataGenerationConfig.GenerateCPEs {
+
+			if cfg.DataGenerationConfig.GenerateCPEs && !hasAuthoritativeCPE(p.CPEs) {
 				// generate CPEs (note: this is excluded from package ID, so is safe to mutate)
 				// we might have binary classified CPE already with the package so we want to append here
-				dictionaryCPEs, ok := cpe.DictionaryFind(p)
+				dictionaryCPEs, ok := cpeutils.DictionaryFind(p)
 				if ok {
 					log.Tracef("used CPE dictionary to find CPEs for %s package %q: %s", p.Type, p.Name, dictionaryCPEs)
 					p.CPEs = append(p.CPEs, dictionaryCPEs...)
 				} else {
-					p.CPEs = append(p.CPEs, cpe.Generate(p)...)
+					p.CPEs = append(p.CPEs, cpeutils.Generate(p)...)
 				}
 			}
 
@@ -155,6 +157,15 @@ func NewPackageTask(cfg CatalogingFactoryConfig, c pkg.Cataloger, tags ...string
 	return NewTask(c.Name(), fn, tags...)
 }
 
+func hasAuthoritativeCPE(cpes []cpe.CPE) bool {
+	for _, c := range cpes {
+		if c.Source != cpe.GeneratedSource {
+			return true
+		}
+	}
+	return false
+}
+
 func prettyName(s string) string {
 	if s == "" {
 		return ""

internal/task/package_task_factory_test.go

@@ -0,0 +1,55 @@
+package task
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/anchore/syft/syft/cpe"
+)
+
+func Test_hasAuthoritativeCPE(t *testing.T) {
+	tests := []struct {
+		name string
+		cpes []cpe.CPE
+		want bool
+	}{
+		{
+			name: "no cpes",
+			cpes: []cpe.CPE{},
+			want: false,
+		},
+		{
+			name: "no authoritative cpes",
+			cpes: []cpe.CPE{
+				{
+					Source: cpe.GeneratedSource,
+				},
+			},
+			want: false,
+		},
+		{
+			name: "has declared (authoritative) cpe",
+			cpes: []cpe.CPE{
+				{
+					Source: cpe.DeclaredSource,
+				},
+			},
+			want: true,
+		},
+		{
+			name: "has lookup (authoritative) cpe",
+			cpes: []cpe.CPE{
+				{
+					Source: cpe.NVDDictionaryLookupSource,
+				},
+			},
+			want: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, hasAuthoritativeCPE(tt.cpes))
+		})
+	}
+}

syft/pkg/cataloger/binary/classifier.go

@@ -274,9 +274,13 @@ func getContents(context matcherContext) ([]byte, error) {
 
 // singleCPE returns a []cpe.CPE with Source: Generated based on the cpe string or panics if the
 // cpe string cannot be parsed into valid CPE Attributes
-func singleCPE(cpeString string) []cpe.CPE {
+func singleCPE(cpeString string, source ...cpe.Source) []cpe.CPE {
+	src := cpe.GeneratedSource
+	if len(source) > 0 {
+		src = source[0]
+	}
 	return []cpe.CPE{
-		cpe.Must(cpeString, cpe.GeneratedSource),
+		cpe.Must(cpeString, src),
 	}
 }
 

syft/pkg/cataloger/binary/classifiers.go

@@ -537,7 +537,7 @@ func DefaultClassifiers() []Classifier {
 			),
 			Package: "curl",
 			PURL:    mustPURL("pkg:generic/curl@version"),
-			CPEs:    singleCPE("cpe:2.3:a:haxx:curl:*:*:*:*:*:*:*:*"),
+			CPEs:    singleCPE("cpe:2.3:a:haxx:curl:*:*:*:*:*:*:*:*", cpe.NVDDictionaryLookupSource),
 		},
 	}
 }