feat(nvd): use go to upload NVD conversion to gcs upon conversion

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

@@ -16,8 +16,8 @@ import (
 
 	"cloud.google.com/go/storage"
 	"github.com/google/osv/vulnfeeds/conversion"
+	"github.com/google/osv/vulnfeeds/gcs-tools"
 	"github.com/google/osv/vulnfeeds/models"
-	"github.com/google/osv/vulnfeeds/upload"
 	"github.com/google/osv/vulnfeeds/utility/logger"
 	"github.com/ossf/osv-schema/bindings/go/osvschema"
 	"google.golang.org/api/iterator"
@@ -92,7 +92,7 @@ func main() {
 		vulnerabilities = append(vulnerabilities, v)
 	}
 
-	upload.Upload(ctx, "OSV files", *uploadToGCS, *outputBucketName, *overridesBucketName, *numWorkers, *osvOutputPath, vulnerabilities, *syncDeletions)
+	gcs.Upload(ctx, "OSV files", *uploadToGCS, *outputBucketName, *overridesBucketName, *numWorkers, *osvOutputPath, vulnerabilities, *syncDeletions)
 }
 
 // extractCVEName extracts the CVE name from a given filename and prefix.

vulnfeeds/cmd/converters/alpine/main.go

@@ -15,8 +15,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/google/osv/vulnfeeds/gcs-tools"
 	"github.com/google/osv/vulnfeeds/models"
-	"github.com/google/osv/vulnfeeds/upload"
 	"github.com/google/osv/vulnfeeds/utility/logger"
 	"github.com/google/osv/vulnfeeds/vulns"
 	"github.com/ossf/osv-schema/bindings/go/osvschema"
@@ -64,7 +64,7 @@ func main() {
 	}
 
 	ctx := context.Background()
-	upload.Upload(ctx, "Alpine CVEs", *uploadToGCS, *outputBucketName, "", *numWorkers, *alpineOutputPath, vulnerabilities, *syncDeletions)
+	gcs.Upload(ctx, "Alpine CVEs", *uploadToGCS, *outputBucketName, "", *numWorkers, *alpineOutputPath, vulnerabilities, *syncDeletions)
 	logger.Info("Alpine CVE conversion succeeded.")
 }
 

vulnfeeds/cmd/converters/cve/nvd-cve-osv/main.go

@@ -2,6 +2,7 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"flag"
 	"fmt"
@@ -14,11 +15,14 @@ import (
 	"slices"
 	"sync"
 
+	"cloud.google.com/go/storage"
 	c "github.com/google/osv/vulnfeeds/conversion"
 	"github.com/google/osv/vulnfeeds/conversion/nvd"
+	gcs "github.com/google/osv/vulnfeeds/gcs-tools"
 	"github.com/google/osv/vulnfeeds/git"
 	"github.com/google/osv/vulnfeeds/models"
 	"github.com/google/osv/vulnfeeds/utility/logger"
+	"github.com/google/osv/vulnfeeds/vulns"
 )
 
 var (
@@ -30,6 +34,9 @@ var (
 	rejectFailed        = flag.Bool("reject-failed", false, "If set, OSV records with a failed conversion outcome will not be generated.")
 	outputMetrics       = flag.Bool("output-metrics", true, "If true, output the metrics information about the conversion")
 	cpuProfile          = flag.String("cpuprofile", "", "Path to write cpu profile to file (default = no output)")
+	uploadToGCS         = flag.Bool("upload-to-gcs", false, "If true, upload to GCS bucket instead of writing to local disk.")
+	outputBucket        = flag.String("output-bucket", "osv-test-cve-osv-conversion", "The GCS bucket to write to.")
+	gcsPrefix           = flag.String("gcs-prefix", "nvd-osv", "The prefix within the GCS bucket.")
 )
 
 func loadCPEDictionary(productToRepo *c.VPRepoCache, f string) error {
@@ -91,12 +98,24 @@ func main() {
 
 	repoTagsCache := &git.RepoTagsCache{}
 
+	var bkt *storage.BucketHandle
+	ctx := context.Background()
+	if *uploadToGCS {
+		client, err := storage.NewClient(ctx)
+		if err != nil {
+			logger.Fatal("Failed to create GCS client", slog.Any("err", err))
+		}
+		defer client.Close()
+		bkt = client.Bucket(*outputBucket)
+		logger.Info("GCS Client and Bucket initialized", slog.String("bucket", *outputBucket))
+	}
+
 	jobs := make(chan models.NVDCVE)
 	var wg sync.WaitGroup
 
 	for range *workers {
 		wg.Add(1)
-		go worker(&wg, jobs, *outDir, vpRepoCache, repoTagsCache)
+		go worker(ctx, &wg, jobs, bkt, *outDir, vpRepoCache, repoTagsCache)
 	}
 
 	for _, cve := range parsed.Vulnerabilities {
@@ -122,7 +141,7 @@ func main() {
 	}
 }
 
-func processCVE(cve models.NVDCVE, vpRepoCache *c.VPRepoCache, repoTagsCache *git.RepoTagsCache) models.ConversionOutcome {
+func processCVE(cve models.NVDCVE, vpRepoCache *c.VPRepoCache, repoTagsCache *git.RepoTagsCache) (*vulns.Vulnerability, *models.ConversionMetrics, models.ConversionOutcome) {
 	metrics := &models.ConversionMetrics{
 		CVEID: cve.ID,
 		CNA:   "nvd",
@@ -131,24 +150,73 @@ func processCVE(cve models.NVDCVE, vpRepoCache *c.VPRepoCache, repoTagsCache *gi
 	metrics.Repos = repos
 
 	var outcome models.ConversionOutcome
+	var vuln *vulns.Vulnerability
+	var finalMetrics *models.ConversionMetrics
 	switch *outFormat {
 	case "OSV":
-		outcome = nvd.CVEToOSV(cve, repos, repoTagsCache, *outDir, metrics, *rejectFailed, *outputMetrics)
+		vuln, finalMetrics, outcome = nvd.CVEToOSV(cve, repos, repoTagsCache, metrics)
 	case "PackageInfo":
 		outcome = nvd.CVEToPackageInfo(cve, repos, repoTagsCache, *outDir, metrics)
+		finalMetrics = metrics
 	}
 
-	return outcome
+	return vuln, finalMetrics, outcome
 }
 
-func worker(wg *sync.WaitGroup, jobs <-chan models.NVDCVE, _ string, vpRepoCache *c.VPRepoCache, repoTagsCache *git.RepoTagsCache) {
+func worker(ctx context.Context, wg *sync.WaitGroup, jobs <-chan models.NVDCVE, bkt *storage.BucketHandle, outDir string, vpRepoCache *c.VPRepoCache, repoTagsCache *git.RepoTagsCache) {
 	defer wg.Done()
 	for cve := range jobs {
-		outcome := processCVE(cve, vpRepoCache, repoTagsCache)
+		vuln, metrics, outcome := processCVE(cve, vpRepoCache, repoTagsCache)
+		cveID := string(cve.ID)
+		if outcome == models.Error {
+			logger.Error("Error generating OSV record", slog.String("cve", cveID), slog.String("outcome", outcome.String()))
+			return // Don't attempt to output files if there was an error
+		}
+
 		if outcome != models.Successful {
-			logger.Info("Failed to generate an OSV record", slog.String("cve", string(cve.ID)), slog.String("outcome", outcome.String()))
+			logger.Info("Failed to generate a successful OSV record", slog.String("cve", cveID), slog.String("outcome", outcome.String()))
+			if *rejectFailed {
+				return // Skip outputting OSV file
+			}
+		} else {
+			logger.Info("Generated OSV record for "+cveID, slog.String("cve", cveID))
+		}
+
+		if *uploadToGCS && bkt != nil {
+			if vuln != nil {
+				if err := gcs.UploadVulnerability(ctx, bkt, *gcsPrefix, vuln.Vulnerability); err != nil {
+					logger.Error("Failed to upload vulnerability", slog.String("cve", vuln.Id), slog.Any("err", err))
+				}
+			}
+			if *outputMetrics && metrics != nil {
+				if err := gcs.UploadMetrics(ctx, bkt, *gcsPrefix, models.CVEID(cveID), metrics); err != nil {
+					logger.Error("Failed to upload metrics", slog.String("cve", cveID), slog.Any("err", err))
+				}
+			}
 		} else {
-			logger.Info("Generated OSV record for "+string(cve.ID), slog.String("cve", string(cve.ID)))
+			// Local file output
+			if vuln != nil {
+				osvFile, err := c.CreateOSVFile(models.CVEID(vuln.Id), outDir)
+				if err != nil {
+					logger.Error("Failed to create OSV file locally", slog.String("cve", vuln.Id), slog.Any("err", err))
+				} else {
+					if err := vuln.ToJSON(osvFile); err != nil {
+						logger.Error("Failed to write OSV file locally", slog.String("cve", vuln.Id), slog.Any("err", err))
+					}
+					osvFile.Close()
+				}
+			}
+			if *outputMetrics && metrics != nil {
+				metricsFile, err := c.CreateMetricsFile(models.CVEID(cveID), outDir)
+				if err != nil {
+					logger.Error("Failed to create metrics file locally", slog.String("cve", cveID), slog.Any("err", err))
+				} else {
+					if err := c.WriteMetricsFile(metrics, metricsFile); err != nil {
+						logger.Error("Failed to write metrics file locally", slog.String("cve", cveID), slog.Any("err", err))
+					}
+					metricsFile.Close()
+				}
+			}
 		}
 	}
 }

vulnfeeds/cmd/converters/cve/nvd-cve-osv/run_cve_to_osv_generation.sh

@@ -63,21 +63,4 @@ for (( YEAR = $(date +%Y) ; YEAR >= ${FIRST_INSCOPE_YEAR} ; YEAR-- )); do
     -exec cp '{}' "${WORK_DIR}/nvd2osv/gcs_stage/" \;
 done
 
-# Copy (and remove any missing) results to GCS bucket, with some sanity
-# checking.
-objs_present=$(gsutil ls "${OSV_OUTPUT_GCS_PATH}" | wc -l)
-objs_deleted=$(gsutil -m rsync -c -n -d "${WORK_DIR}/nvd2osv/gcs_stage" "${OSV_OUTPUT_GCS_PATH}" 2>&1 | grep "Would remove" | wc -l)
-
-threshold=$(echo "scale=2; ${objs_present} * (${SAFETY_THRESHOLD_PCT:-2} / 100)" | bc)
-
-# Bash can't deal with floats
-if (( $(echo "${objs_deleted} > ${threshold}" | bc -l) )); then
-  echo "Warning. Unexpectedly high (${objs_deleted}) number of CVE records would be deleted!" >> /dev/stderr
-  gsutil -m rsync -c -n -d "${WORK_DIR}/nvd2osv/gcs_stage" "${OSV_OUTPUT_GCS_PATH}" 2>&1 | grep "Would remove" >> /dev/stderr
-  # TODO: add back in once nvd-mirror issue fixed: exit 1
-fi
-
-echo "Copying NVD CVE records successfully converted to GCS bucket"
-gsutil -q -m rsync -c "${WORK_DIR}/nvd2osv/gcs_stage" "${OSV_OUTPUT_GCS_PATH}"
-
 echo "Conversion run complete"

vulnfeeds/cmd/converters/debian/main.go

@@ -15,8 +15,8 @@ import (
 	"strings"
 
 	"github.com/google/osv/vulnfeeds/faulttolerant"
+	"github.com/google/osv/vulnfeeds/gcs-tools"
 	"github.com/google/osv/vulnfeeds/models"
-	"github.com/google/osv/vulnfeeds/upload"
 	"github.com/google/osv/vulnfeeds/utility/logger"
 	"github.com/google/osv/vulnfeeds/vulns"
 	"github.com/ossf/osv-schema/bindings/go/osvschema"
@@ -70,7 +70,7 @@ func main() {
 	}
 
 	ctx := context.Background()
-	upload.Upload(ctx, "Debian CVEs", *uploadToGCS, *outputBucketName, "", *numWorkers, *debianOutputPath, vulnerabilities, *syncDeletions)
+	gcs.Upload(ctx, "Debian CVEs", *uploadToGCS, *outputBucketName, "", *numWorkers, *debianOutputPath, vulnerabilities, *syncDeletions)
 	logger.Info("Debian CVE conversion succeeded.")
 }
 

vulnfeeds/conversion/nvd/converter.go

@@ -24,21 +24,17 @@ var ErrNoRanges = errors.New("no ranges")
 
 var ErrUnresolvedFix = errors.New("fixes not resolved to commits")
 
-// CVEToOSV Takes an NVD CVE record and outputs an OSV file in the specified directory.
-func CVEToOSV(cve models.NVDCVE, repos []string, cache *git.RepoTagsCache, directory string, metrics *models.ConversionMetrics, rejectFailed bool, outputMetrics bool) models.ConversionOutcome {
+// CVEToOSV Takes an NVD CVE record and returns an OSV Vulnerability object, ConversionMetrics, and the outcome.
+func CVEToOSV(cve models.NVDCVE, repos []string, cache *git.RepoTagsCache, metrics *models.ConversionMetrics) (*vulns.Vulnerability, *models.ConversionMetrics, models.ConversionOutcome) {
 	CPEs := c.CPEs(cve)
 	metrics.CPEs = CPEs
 	// The vendor name and product name are used to construct the output `vulnDir` below, so need to be set to *something* to keep the output tidy.
-	maybeVendorName := "ENOCPE"
-	maybeProductName := "ENOCPE"
 
 	if len(CPEs) > 0 {
-		CPE, err := c.ParseCPE(CPEs[0]) // For naming the subdirectory used for output.
-		maybeVendorName = CPE.Vendor
-		maybeProductName = CPE.Product
+		_, err := c.ParseCPE(CPEs[0]) // For naming the subdirectory used for output.
 		if err != nil {
 			metrics.AddNote("Can't generate an OSV record without valid CPE data")
-			return models.ConversionUnknown
+			return nil, metrics, models.ConversionUnknown
 		}
 	}
 
@@ -52,9 +48,7 @@ func CVEToOSV(cve models.NVDCVE, repos []string, cache *git.RepoTagsCache, direc
 	// If there are no repos, there are no commits from the refs either
 	if len(cpeRanges) == 0 && len(repos) == 0 {
 		metrics.SetOutcome(models.NoRepos)
-		outputFiles(v, directory, maybeVendorName, maybeProductName, metrics, rejectFailed, outputMetrics)
-
-		return models.NoRepos
+		return v, metrics, models.NoRepos
 	}
 
 	successfulRepos := make(map[string]bool)
@@ -67,15 +61,13 @@ func CVEToOSV(cve models.NVDCVE, repos []string, cache *git.RepoTagsCache, direc
 		affected := MergeRangesAndCreateAffected(resolvedRanges, cpeRanges, nil, nil, metrics)
 		v.Affected = append(v.Affected, affected)
 		// Exit early
-		outputFiles(v, directory, maybeVendorName, maybeProductName, metrics, rejectFailed, outputMetrics)
-
-		return models.NoRepos
+		return v, metrics, models.NoRepos
 	}
 
 	// If we have ranges, try to resolve them
 	r, un, sR := processRanges(cpeRanges, repos, metrics, cache, models.VersionSourceCPE)
 	if metrics.Outcome == models.Error {
-		return models.Error
+		return nil, metrics, models.Error
 	}
 	resolvedRanges = append(resolvedRanges, r...)
 	unresolvedRanges = append(unresolvedRanges, un...)
@@ -105,7 +97,7 @@ func CVEToOSV(cve models.NVDCVE, repos []string, cache *git.RepoTagsCache, direc
 		}
 		r, un, sR := processRanges(textRanges, repos, metrics, cache, models.VersionSourceDescription)
 		if metrics.Outcome == models.Error {
-			return models.Error
+			return nil, metrics, models.Error
 		}
 		resolvedRanges = append(resolvedRanges, r...)
 		unresolvedRanges = append(unresolvedRanges, un...)
@@ -115,7 +107,7 @@ func CVEToOSV(cve models.NVDCVE, repos []string, cache *git.RepoTagsCache, direc
 	}
 
 	if len(resolvedRanges) == 0 && len(commits) == 0 {
-		metrics.AddNote("No ranges detected for %q", maybeProductName)
+		metrics.AddNote("No ranges detected")
 		metrics.SetOutcome(models.NoRanges)
 	}
 
@@ -124,13 +116,11 @@ func CVEToOSV(cve models.NVDCVE, repos []string, cache *git.RepoTagsCache, direc
 	affected := MergeRangesAndCreateAffected(resolvedRanges, unresolvedRanges, commits, keys, metrics)
 	v.Affected = append(v.Affected, affected)
 
-	if metrics.Outcome == models.Error || (!outputMetrics && rejectFailed && metrics.Outcome != models.Successful) {
-		return metrics.Outcome
+	if metrics.Outcome == models.Error {
+		return nil, metrics, metrics.Outcome
 	}
 
-	outputFiles(v, directory, maybeVendorName, maybeProductName, metrics, rejectFailed, outputMetrics)
-
-	return metrics.Outcome
+	return v, metrics, metrics.Outcome
 }
 
 // CVEToPackageInfo takes an NVD CVE record and outputs a PackageInfo struct in a file in the specified directory.
@@ -448,52 +438,6 @@ func convertCommitToEvent(commit models.AffectedCommit) *osvschema.Event {
 	return nil
 }
 
-// outputFiles writes the OSV vulnerability record and conversion metrics to files in the specified directory.
-// It creates the necessary subdirectories based on the vendor and product names and handles whether or not
-// the files should be written based on the rejectFailed and outputMetrics flags.
-//
-// Arguments:
-//   - v: The OSV Vulnerability object to be written to a file.
-//   - dir: The base directory where the output files should be created.
-//   - vendor: The vendor name used to create the subdirectory.
-//   - product: The product name used to create the subdirectory.
-//   - metrics: A pointer to ConversionMetrics to be written to a metrics file.
-//   - rejectFailed: A boolean indicating whether to skip writing the OSV file if the conversion was not successful.
-//   - outputMetrics: A boolean indicating whether to write the metrics file.
-func outputFiles(v *vulns.Vulnerability, dir string, vendor string, product string, metrics *models.ConversionMetrics, rejectFailed bool, outputMetrics bool) {
-	cveID := v.Id
-	vulnDir := filepath.Join(dir, vendor, product)
-
-	if err := os.MkdirAll(vulnDir, 0755); err != nil {
-		logger.Info("Failed to create directory "+vulnDir, slog.String("cve", cveID), slog.String("path", vulnDir), slog.Any("err", err))
-	}
-
-	if metrics.Outcome == models.Error {
-		return
-	}
-
-	if !rejectFailed || metrics.Outcome == models.Successful {
-		osvFile, errCVE := c.CreateOSVFile(models.CVEID(cveID), vulnDir)
-		if errCVE != nil {
-			logger.Fatal("File failed to be created for CVE", slog.String("cve", cveID))
-		}
-		if err := v.ToJSON(osvFile); err != nil {
-			logger.Error("Failed to write", slog.Any("err", err))
-		}
-		osvFile.Close()
-	}
-	if outputMetrics {
-		metricsFile, errMetrics := c.CreateMetricsFile(models.CVEID(cveID), vulnDir)
-		if errMetrics != nil {
-			logger.Fatal("File failed to be created for CVE", slog.String("cve", cveID))
-		}
-		if err := c.WriteMetricsFile(metrics, metricsFile); err != nil {
-			logger.Error("Failed to write metrics", slog.Any("err", err))
-		}
-		metricsFile.Close()
-	}
-}
-
 // processRanges attempts to resolve the given ranges to commits and updates the metrics accordingly.
 func processRanges(ranges []*osvschema.Range, repos []string, metrics *models.ConversionMetrics, cache *git.RepoTagsCache, source models.VersionSource) ([]*osvschema.Range, []*osvschema.Range, []string) {
 	if len(ranges) == 0 {

vulnfeeds/conversion/nvd/converter_test.go

@@ -65,7 +65,7 @@ func TestCVEToOSV_429(t *testing.T) {
 	cache := &git.RepoTagsCache{}
 	outDir := t.TempDir()
 
-	outcome := CVEToOSV(cve, []string{"https://github.com/foo/bar"}, cache, outDir, metrics, false, false)
+	_, _, outcome := CVEToOSV(cve, []string{"https://github.com/foo/bar"}, cache, metrics)
 
 	// It should fail because of the 429 error causing unresolved fixes
 	if outcome != models.Error {

vulnfeeds/upload/cveworker.go → vulnfeeds/gcs-tools/cveworker.go

@@ -1,5 +1,5 @@
-// Package upload handles allocating workers to intelligently uploading OSV records to a bucket
-package upload
+// Package gcs handles allocating workers to intelligently uploading OSV records to a bucket
+package gcs
 
 import (
 	"bytes"

vulnfeeds/upload/cveworker_test.go → vulnfeeds/gcs-tools/cveworker_test.go

@@ -1,4 +1,4 @@
-package upload
+package gcs
 
 import (
 	"bytes"