refactor(vulnfeeds): Uncouple Alpine vulns (#3954)

Author: jess-lowe

deployment/clouddeploy/gke-workers/base/alpine-cve-convert.yaml

@@ -12,6 +12,10 @@ spec:
       activeDeadlineSeconds: 3600
       template:
         spec:
+          tolerations:
+          - key: workloadType
+            operator: Equal
+            value: highend
           containers:
           - name: alpine-cve-convert
             image: alpine-cve-convert
@@ -20,11 +24,11 @@ spec:
               privileged: true
             resources:
               requests:
-                cpu: "1"
-                memory: "1G"
+                cpu: "2"
+                memory: "16G"
               limits:
-                cpu: "1"
-                memory: "2G"
+                cpu: "4"
+                memory: "16G"
           restartPolicy: Never
           volumes:
             - name: "ssd"

deployment/clouddeploy/gke-workers/environments/oss-vdb-test/alpine-cve-convert.yaml

@@ -14,3 +14,7 @@ spec:
               value: oss-vdb-test
             - name: OUTPUT_GCS_BUCKET
               value: osv-test-cve-osv-conversion
+            - name: INPUT_GCS_BUCKET
+              value: osv-test-cve-osv-conversion
+            - name: NUM_WORKERS
+              value: '256'

deployment/clouddeploy/gke-workers/environments/oss-vdb/alpine-cve-convert.yaml

@@ -14,3 +14,7 @@ spec:
               value: oss-vdb
             - name: OUTPUT_GCS_BUCKET
               value: cve-osv-conversion
+            - name: INPUT_GCS_BUCKET
+              value: cve-osv-conversion
+            - name: NUM_WORKERS
+              value: '256'

vulnfeeds/cmd/alpine/Dockerfile

@@ -22,13 +22,13 @@ COPY ./go.sum /src/go.sum
 RUN go mod download
 
 COPY ./ /src/
-RUN go build -o alpine-osv ./cmd/alpine/
+RUN go build -o alpine ./cmd/alpine/
 
 
 FROM gcr.io/google.com/cloudsdktool/google-cloud-cli:alpine@sha256:cdac858d976cb0e6bfdc3288fee5a0a7bf6348a009089be130b2009e28463c52
 
 WORKDIR /root/
-COPY --from=GO_BUILD /src/alpine-osv ./
+COPY --from=GO_BUILD /src/alpine ./
 COPY ./cmd/alpine/run_alpine_convert.sh ./
 
 ENTRYPOINT ["/root/run_alpine_convert.sh"]

vulnfeeds/cmd/alpine/main.go

@@ -2,26 +2,34 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"flag"
 	"fmt"
 	"io"
 	"log/slog"
 	"net/http"
 	"os"
-	"path"
 	"regexp"
+	"sort"
 	"strings"
+	"sync"
+	"time"
 
+	"cloud.google.com/go/storage"
+	"github.com/google/osv/vulnfeeds/cves"
 	"github.com/google/osv/vulnfeeds/models"
 	"github.com/google/osv/vulnfeeds/utility/logger"
 	"github.com/google/osv/vulnfeeds/vulns"
+	"github.com/ossf/osv-schema/bindings/go/osvschema"
 )
 
 const (
 	alpineURLBase           = "https://secdb.alpinelinux.org/%s/main.json"
 	alpineIndexURL          = "https://secdb.alpinelinux.org/"
-	alpineOutputPathDefault = "parts/alpine"
+	alpineOutputPathDefault = "alpine"
+	defaultCvePath          = "cve_jsons"
+	outputBucketDefault     = "osv-test-cve-osv-conversion"
 )
 
 func main() {
@@ -31,15 +39,51 @@ func main() {
 		"alpineOutput",
 		alpineOutputPathDefault,
 		"path to output general alpine affected package information")
+	outputBucketName := flag.String("output_bucket", outputBucketDefault, "The GCS bucket to write to.")
+	numWorkers := flag.Int("num_workers", 64, "Number of workers to process records")
+	uploadToGCS := flag.Bool("uploadToGCS", false, "If true, do not write to GCS bucket and instead write to local disk.")
 	flag.Parse()
 
 	err := os.MkdirAll(*alpineOutputPath, 0755)
 	if err != nil {
 		logger.Fatal("Can't create output path", slog.Any("err", err))
 	}
 
+	allCVEs := vulns.LoadAllCVEs(defaultCvePath)
 	allAlpineSecDB := getAlpineSecDBData()
-	generateAlpineOSV(allAlpineSecDB, *alpineOutputPath)
+	osvVulnerabilities := generateAlpineOSV(allAlpineSecDB, allCVEs)
+
+	ctx := context.Background()
+	var bkt *storage.BucketHandle
+	if *uploadToGCS {
+		storageClient, err := storage.NewClient(ctx)
+		if err != nil {
+			logger.Fatal("Failed to create storage client", slog.Any("err", err))
+		}
+		bkt = storageClient.Bucket(*outputBucketName)
+	}
+	var wg sync.WaitGroup
+	vulnChan := make(chan *vulns.Vulnerability)
+
+	for range *numWorkers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			vulns.Worker(ctx, vulnChan, bkt, *alpineOutputPath)
+		}()
+	}
+
+	for _, v := range osvVulnerabilities {
+		if len(v.Affected) == 0 {
+			logger.Warn(fmt.Sprintf("Skipping %s as no affected versions found.", v.ID), slog.String("id", v.ID))
+			continue
+		}
+		vulnChan <- v
+	}
+
+	close(vulnChan)
+	wg.Wait()
+	logger.Info("Alpine CVE conversion succeeded.")
 }
 
 // getAllAlpineVersions gets all available version name in alpine secdb
@@ -55,7 +99,7 @@ func getAllAlpineVersions() []string {
 		logger.Fatal("Failed to get alpine index page", slog.Any("err", err))
 	}
 
-	exp := regexp.MustCompile("href=\"(v[\\d.]*)/\"")
+	exp := regexp.MustCompile(`href="(v[\d.]*)/"`)
 
 	searchRes := exp.FindAllStringSubmatch(buf.String(), -1)
 	alpineVersions := make([]string, 0, len(searchRes))
@@ -87,8 +131,9 @@ func getAlpineSecDBData() map[string][]VersionAndPkg {
 					cveID = strings.Split(cveID, " ")[0]
 
 					if !validVersion(version) {
-						logger.Warn("Invalid alpine version",
-							slog.String("version", version),
+						logger.Warn(fmt.Sprintf("[%s] Invalid alpine version: '%s', on package: '%s', and alpine version: '%s'",
+							cveID, version, pkg.Pkg.Name,
+							alpineVer), slog.String("version", version),
 							slog.String("package", pkg.Pkg.Name),
 							slog.String("alpine_version", alpineVer),
 						)
@@ -111,9 +156,53 @@ func getAlpineSecDBData() map[string][]VersionAndPkg {
 }
 
 // generateAlpineOSV generates the generic PackageInfo package from the information given by alpine advisory
-func generateAlpineOSV(allAlpineSecDb map[string][]VersionAndPkg, alpineOutputPath string) {
-	for cveID, verPkgs := range allAlpineSecDb {
-		pkgInfos := make([]vulns.PackageInfo, 0, len(verPkgs))
+func generateAlpineOSV(allAlpineSecDb map[string][]VersionAndPkg, allCVEs map[cves.CVEID]cves.Vulnerability) (osvVulnerabilities []*vulns.Vulnerability) {
+	cveIDs := make([]string, 0, len(allAlpineSecDb))
+	for cveID := range allAlpineSecDb {
+		cveIDs = append(cveIDs, cveID)
+	}
+	sort.Strings(cveIDs)
+
+	for _, cveID := range cveIDs {
+		verPkgs := allAlpineSecDb[cveID]
+		sort.Slice(verPkgs, func(i, j int) bool {
+			if verPkgs[i].Pkg != verPkgs[j].Pkg {
+				return verPkgs[i].Pkg < verPkgs[j].Pkg
+			}
+			if verPkgs[i].AlpineVer != verPkgs[j].AlpineVer {
+				return verPkgs[i].AlpineVer < verPkgs[j].AlpineVer
+			}
+
+			return verPkgs[i].Ver < verPkgs[j].Ver
+		})
+		cve, ok := allCVEs[cves.CVEID(cveID)]
+		var published time.Time
+		var details string
+		if ok {
+			published = cve.CVE.Published.Time
+			if len(cve.CVE.Descriptions) > 0 {
+				details = cve.CVE.Descriptions[0].Value
+			}
+		} else {
+			// TODO: add support for non-CVE reports
+			logger.Warn(fmt.Sprintf("CVE %s not found in cve_jsons", cveID), slog.String("cveID", cveID))
+			continue
+		}
+
+		v := &vulns.Vulnerability{
+			Vulnerability: osvschema.Vulnerability{
+				ID:        "ALPINE-" + cveID,
+				Upstream:  []string{cveID},
+				Published: published,
+				Details:   details,
+				References: []osvschema.Reference{
+					{
+						Type: "ADVISORY",
+						URL:  "https://security.alpinelinux.org/vuln/" + cveID,
+					},
+				},
+			},
+		}
 
 		for _, verPkg := range verPkgs {
 			pkgInfo := vulns.PackageInfo{
@@ -124,23 +213,17 @@ func generateAlpineOSV(allAlpineSecDb map[string][]VersionAndPkg, alpineOutputPa
 				Ecosystem: "Alpine:" + verPkg.AlpineVer,
 				PURL:      "pkg:apk/alpine/" + verPkg.Pkg + "?arch=source",
 			}
-			pkgInfos = append(pkgInfos, pkgInfo)
+			v.AddPkgInfo(pkgInfo)
 		}
 
-		file, err := os.OpenFile(path.Join(alpineOutputPath, cveID+".alpine.json"), os.O_CREATE|os.O_RDWR, 0644)
-		if err != nil {
-			logger.Fatal("Failed to create/write osv output file", slog.Any("err", err))
-		}
-		encoder := json.NewEncoder(file)
-		encoder.SetIndent("", "  ")
-		err = encoder.Encode(&pkgInfos)
-		if err != nil {
-			logger.Fatal("Failed to encode package info output file", slog.Any("err", err))
+		if len(v.Affected) == 0 {
+			logger.Warn(fmt.Sprintf("Skipping %s as no affected versions found.", v.ID), slog.String("cveID", cveID))
+			continue
 		}
-		_ = file.Close()
+		osvVulnerabilities = append(osvVulnerabilities, v)
 	}
 
-	logger.Info("Finished")
+	return osvVulnerabilities
 }
 
 // downloadAlpine downloads Alpine SecDB data from their API

vulnfeeds/cmd/alpine/run_alpine_convert.sh

@@ -8,13 +8,20 @@
 
 set -e
 
-OSV_PARTS_OUTPUT="parts/alpine"
-OUTPUT_BUCKET="${OUTPUT_GCS_BUCKET:=cve-osv-conversion}"
+OSV_OUTPUT_PATH="alpine"
+INPUT_BUCKET="${INPUT_GCS_BUCKET:=osv-test-cve-osv-conversion}"
+OUTPUT_BUCKET="${OUTPUT_GCS_BUCKET:=osv-test-cve-osv-conversion}"
+CVE_OUTPUT="cve_jsons/"
+WORKERS="${NUM_WORKERS:=256}"
 
-echo "Setup initial directories"
-rm -rf $OSV_PARTS_OUTPUT && mkdir -p $OSV_PARTS_OUTPUT
 
-./alpine-osv
-echo "Begin Syncing with cloud"
-gsutil -q -m rsync -c -d $OSV_PARTS_OUTPUT "gs://$OUTPUT_BUCKET/$OSV_PARTS_OUTPUT"
-echo "Successfully synced with cloud"
+echo "Setup initial directories ${OSV_OUTPUT_PATH}"
+rm -rf $OSV_OUTPUT_PATH && mkdir -p $OSV_OUTPUT_PATH
+rm -rf $CVE_OUTPUT && mkdir -p $CVE_OUTPUT
+
+echo "Begin syncing NVD data from GCS bucket ${INPUT_BUCKET}"
+gcloud --no-user-output-enabled storage -q cp "gs://${INPUT_BUCKET}/nvd/*-????.json" "${CVE_OUTPUT}"
+echo "Successfully synced from GCS bucket"
+
+./alpine -output_bucket "$OUTPUT_BUCKET" -output_path "$OSV_OUTPUT_PATH" -num_workers "$WORKERS" -uploadToGCS
+echo "Successfully converted and uploaded to cloud"

vulnfeeds/cmd/combine-to-osv/main.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"flag"
 	"log/slog"
-	"net/url"
 	"os"
 	"path"
 	"strings"
@@ -14,7 +13,6 @@ import (
 	"github.com/google/osv/vulnfeeds/cves"
 	"github.com/google/osv/vulnfeeds/utility/logger"
 	"github.com/google/osv/vulnfeeds/vulns"
-	"github.com/ossf/osv-schema/bindings/go/osvschema"
 )
 
 const (
@@ -23,9 +21,8 @@ const (
 	defaultOSVOutputPath  = "osv_output"
 	defaultCVEListPath    = "."
 
-	alpineEcosystem          = "Alpine"
-	debianEcosystem          = "Debian"
-	alpineSecurityTrackerURL = "https://security.alpinelinux.org/vuln"
+	alpineEcosystem = "Alpine"
+	debianEcosystem = "Debian"
 )
 
 func main() {
@@ -165,17 +162,12 @@ func combineIntoOSV(loadedCves map[cves.CVEID]cves.Vulnerability, allParts map[c
 			}
 		}
 
-		addedAlpineURL := false
 		for _, pkgInfo := range allParts[cveID] {
-			// skip debian parts, but still write out the CVEs.
-			if strings.HasPrefix(pkgInfo.Ecosystem, debianEcosystem) {
+			// skip debian and alpine parts, but still write out the CVEs.
+			if strings.HasPrefix(pkgInfo.Ecosystem, debianEcosystem) || strings.HasPrefix(pkgInfo.Ecosystem, alpineEcosystem) {
 				continue
 			}
 			convertedCve.AddPkgInfo(pkgInfo)
-			if strings.HasPrefix(pkgInfo.Ecosystem, alpineEcosystem) && !addedAlpineURL {
-				addReference(string(cveID), alpineEcosystem, convertedCve)
-				addedAlpineURL = true
-			}
 		}
 
 		cveModified := convertedCve.Modified
@@ -207,17 +199,3 @@ func writeOSVFile(osvData map[cves.CVEID]*vulns.Vulnerability, osvOutputPath str
 
 	logger.Info("Successfully written OSV files", slog.Int("count", len(osvData)))
 }
-
-// addReference adds the related security tracker URL to a given vulnerability's references
-func addReference(cveID string, ecosystem string, convertedCve *vulns.Vulnerability) {
-	securityReference := osvschema.Reference{Type: osvschema.ReferenceAdvisory}
-	if ecosystem == alpineEcosystem {
-		securityReference.URL, _ = url.JoinPath(alpineSecurityTrackerURL, cveID)
-	}
-
-	if securityReference.URL == "" {
-		return
-	}
-
-	convertedCve.References = append(convertedCve.References, securityReference)
-}

vulnfeeds/cmd/combine-to-osv/main_test.go

@@ -99,32 +99,6 @@ func TestCombineIntoOSV(t *testing.T) {
 	if actualCombined != expectedCombined {
 		t.Errorf("Expected %d in combination, got %d: %#v", expectedCombined, actualCombined, combinedOSV)
 	}
-	for cve := range cveStuff {
-		if len(combinedOSV[cve].Affected) != len(allParts[cve]) {
-			t.Errorf("Affected lengths for %s do not match", cve)
-		}
-
-		found := false
-		switch cve {
-		case "CVE-2022-33745":
-			for _, reference := range combinedOSV[cve].References {
-				if reference.Type == "ADVISORY" &&
-					reference.URL == "https://security.alpinelinux.org/vuln/CVE-2022-33745" {
-					found = true
-				}
-			}
-		case "CVE-2022-32746":
-			for _, reference := range combinedOSV[cve].References {
-				if reference.Type == "ADVISORY" &&
-					reference.URL == "https://security.alpinelinux.org/vuln/CVE-2022-32746" {
-					found = true
-				}
-			}
-		}
-		if !found {
-			t.Errorf("%s doesn't have all expected references", cve)
-		}
-	}
 }
 func TestGetModifiedTime(t *testing.T) {
 	_, err := getModifiedTime("../../test_data/parts/debian/CVE-2016-1585.debian.json")