using osv modified csv file to get the correct modified vulnerabilities, using osv for malicious packages, solves: #1834

Author: timbastin

transformer/osv_transformer.go

@@ -286,7 +286,7 @@ func affectedComponentsFromGitRange(affected dtos.Affected) []models.AffectedCom
 }
 
 // MaliciousAffectedComponentFromOSV converts OSV data to MaliciousAffectedComponent entries
-func MaliciousAffectedComponentFromOSV(osv dtos.OSV, maliciousPackageID string) []models.MaliciousAffectedComponent {
+func MaliciousAffectedComponentFromOSV(osv *dtos.OSV, maliciousPackageID string) []models.MaliciousAffectedComponent {
 	affectedComponents := make([]models.MaliciousAffectedComponent, 0)
 	for _, affected := range osv.Affected {
 		for _, c := range affectedComponentsFromAffected(affected) {

vulndb/gob.go

@@ -222,7 +222,7 @@ func readGobFile(path string, out any) error {
 
 var batchSize = 5_000
 
-func readGobFileStream[T any, Transformed any](ctx context.Context, path string, out chan<- Transformed, transformer func([]T) Transformed) error {
+func readGobFileStream[T any](ctx context.Context, path string, handleItems func([]T) error) error {
 	fd, err := os.Open(path)
 	if err != nil {
 		return fmt.Errorf("could not open gob file %s: %w", path, err)
@@ -241,16 +241,19 @@ func readGobFileStream[T any, Transformed any](ctx context.Context, path string,
 		batch = append(batch, item)
 		if len(batch) == batchSize {
 			select {
-			case out <- transformer(batch):
 			case <-ctx.Done():
 				return ctx.Err()
+			default:
+				handleItems(batch)
 			}
 			batch = batch[:0]
 		}
 	}
 	if len(batch) > 0 {
 		select {
-		case out <- transformer(batch):
+		default:
+			handleItems(batch)
+			return nil
 		case <-ctx.Done():
 			return ctx.Err()
 		}

vulndb/integrity.go

@@ -184,13 +184,13 @@ func calculateTotalIntegrityInformation(ctx context.Context, tx pgx.Tx) ([]table
 			       md5(string_agg(row_hash, '' ORDER BY id)) AS checksum
 			FROM (
 				SELECT id, md5(coalesce(id, '\0') || '|' || coalesce(modified::text, '\0')) AS row_hash
-				FROM malicious_packages WHERE id NOT LIKE 'FAKE-TEST-%'
+				FROM malicious_packages WHERE id NOT LIKE 'MAL-FAKE-TEST-%'
 			) sub
 		),
 		malicious_affected_components_integrity AS (
 			SELECT 'malicious_affected_components' AS table_name, count(*) AS row_count,
 			       md5(string_agg(id::text, '' ORDER BY id)) AS checksum
-			FROM malicious_affected_components WHERE malicious_package_id NOT LIKE 'FAKE-TEST-%'
+			FROM malicious_affected_components WHERE malicious_package_id NOT LIKE 'MAL-FAKE-TEST-%'
 		)
 		SELECT table_name, row_count, checksum FROM cves_integrity
 		UNION ALL SELECT table_name, row_count, checksum FROM cve_relationships_integrity

vulndb/malicious_packages_checker.go

@@ -16,16 +16,11 @@
 package vulndb
 
 import (
-	"archive/tar"
-	"compress/gzip"
 	"context"
-	"encoding/json"
 	"fmt"
-	"io"
 	"log/slog"
 	"net/http"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/jackc/pgx/v5"
@@ -50,12 +45,11 @@ type MaliciousPackageChecker struct {
 	httpClient *http.Client
 }
 
-type malRow struct {
+type malRows struct {
 	pkgs  []models.MaliciousPackage
 	comps []models.MaliciousAffectedComponent
 }
 
-
 func NewMaliciousPackageChecker(
 	repository *repositories.MaliciousPackageRepository,
 ) (*MaliciousPackageChecker, error) {
@@ -68,145 +62,6 @@ func NewMaliciousPackageChecker(
 
 // FetchAll downloads the malicious packages archive and returns all parsed packages
 // and affected components without touching the database.
-func (c *MaliciousPackageChecker) FetchAll(ctx context.Context) ([]models.MaliciousPackage, []models.MaliciousAffectedComponent, error) {
-	slog.Info("Downloading malicious packages archive", "url", c.repoURL)
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.repoURL, nil)
-	if err != nil {
-		return nil, nil, fmt.Errorf("failed to create download request: %w", err)
-	}
-	resp, err := c.httpClient.Do(req)
-	if err != nil {
-		return nil, nil, fmt.Errorf("failed to download archive: %w", err)
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode != http.StatusOK {
-		return nil, nil, fmt.Errorf("failed to download archive: HTTP %d", resp.StatusCode)
-	}
-
-	gzr, err := gzip.NewReader(resp.Body)
-	if err != nil {
-		return nil, nil, fmt.Errorf("failed to create gzip reader: %w", err)
-	}
-	defer gzr.Close()
-
-	tr := tar.NewReader(gzr)
-	ecosystems := []string{"npm", "go", "maven", "pypi", "crates.io"}
-
-	processWG := &sync.WaitGroup{}
-	collectWG := &sync.WaitGroup{}
-
-	fileJobs := make(chan []byte, malPkgNumOfGoRoutines*20)
-	resultJobs := make(chan processingResults, BatchSize*2)
-
-	for range malPkgNumOfGoRoutines {
-		processWG.Add(1)
-		go processMaliciousPackageFile(processWG, fileJobs, resultJobs)
-	}
-
-	var (
-		packages   []models.MaliciousPackage
-		components []models.MaliciousAffectedComponent
-		mu         sync.Mutex
-	)
-	collectWG.Add(1)
-	go func() {
-		defer collectWG.Done()
-		for r := range resultJobs {
-			// pre-compute component IDs so they are stable in the gob file
-			for i := range r.AffectedComponents {
-				if r.AffectedComponents[i].ID == "" {
-					r.AffectedComponents[i].ID = r.AffectedComponents[i].CalculateHash()
-				}
-			}
-			mu.Lock()
-			packages = append(packages, r.Package)
-			components = append(components, r.AffectedComponents...)
-			mu.Unlock()
-		}
-	}()
-
-	for {
-		header, err := tr.Next()
-		if err != nil {
-			if err == io.EOF {
-				break
-			}
-			return nil, nil, fmt.Errorf("failed to read tar: %w", err)
-		}
-		if !strings.HasSuffix(header.Name, ".json") || header.Typeflag != tar.TypeReg {
-			continue
-		}
-		isTarget := false
-		for _, eco := range ecosystems {
-			if strings.Contains(header.Name, "/osv/malicious/"+eco+"/") {
-				isTarget = true
-				break
-			}
-		}
-		if !isTarget {
-			continue
-		}
-		data, err := io.ReadAll(tr)
-		if err != nil {
-			slog.Debug("Failed to read file from tar", "name", header.Name, "error", err)
-			continue
-		}
-		fileJobs <- data
-	}
-
-	close(fileJobs)
-	processWG.Wait()
-	close(resultJobs)
-	collectWG.Wait()
-
-	slog.Info("Fetched malicious packages", "packages", len(packages), "components", len(components))
-	return packages, components, nil
-}
-
-type processingResults struct {
-	Package            models.MaliciousPackage
-	AffectedComponents []models.MaliciousAffectedComponent
-}
-
-// this function grabs json file contents from the jobs channel and builds the package as well as the affected components from it. These are then sent to the db worker function
-func processMaliciousPackageFile(waitGroup *sync.WaitGroup, jobs chan []byte, results chan processingResults) {
-	defer waitGroup.Done()
-	for data := range jobs {
-		var entry dtos.OSV
-		if err := json.Unmarshal(data, &entry); err != nil {
-			slog.Debug("Failed to unmarshal JSON", "error", err)
-			continue
-		}
-
-		if entry.ID == "" {
-			slog.Warn("Skipping malicious package with empty ID", "summary", entry.Summary)
-			continue
-		}
-
-		if len(entry.Affected) == 0 {
-			continue
-		}
-
-		// Create malicious package record
-		pkg := models.MaliciousPackage{
-			ID:        entry.ID,
-			Summary:   entry.Summary,
-			Details:   entry.Details,
-			Published: entry.Published,
-			Modified:  entry.Modified,
-		}
-
-		// Create affected components
-		components := transformer.MaliciousAffectedComponentFromOSV(entry, entry.ID)
-		// send both as a job to the db writer function
-		results <- processingResults{
-			Package:            pkg,
-			AffectedComponents: components,
-		}
-	}
-}
-
 func buildFakePackages() ([]models.MaliciousPackage, []models.MaliciousAffectedComponent) {
 	testPackages := map[string][]string{
 		"npm":       {"fake-malicious-npm-package", "@fake-org/malicious-package"},
@@ -223,8 +78,8 @@ func buildFakePackages() ([]models.MaliciousPackage, []models.MaliciousAffectedC
 	for ecosystem, pkgNames := range testPackages {
 		for _, pkgName := range pkgNames {
 			normalizedPkgName := strings.NewReplacer("/", "-", "@", "-", ":", "-", ".", "-").Replace(pkgName)
-			fakeID := fmt.Sprintf("FAKE-TEST-%s-%s", strings.ToUpper(ecosystem), strings.ToUpper(normalizedPkgName))
-			fakeEntry := dtos.OSV{
+			fakeID := fmt.Sprintf("MAL-FAKE-TEST-%s-%s", strings.ToUpper(ecosystem), strings.ToUpper(normalizedPkgName))
+			fakeEntry := &dtos.OSV{
 				ID:      fakeID,
 				Summary: fmt.Sprintf("Fake malicious %s package for testing", ecosystem),
 				Details: "This is a fake malicious package entry used for testing the dependency proxy",