fix: Race condition in concurrent analyzer

guard/guard.go

@@ -206,13 +206,19 @@ func (g *packageManagerGuard) concurrentAnalyzePackages(ctx context.Context,
 		}()
 	}
 
+	// Queue all packages for analysis
 	for _, pkg := range packages {
 		jobs <- pkg
 	}
 	close(jobs)
 
 	analysisResults := []*analyzer.PackageVersionAnalysisResult{}
+
+	// We must wait for the results go routine to collect all results
+	rwg := sync.WaitGroup{}
+	rwg.Add(1)
 	go func() {
+		defer rwg.Done()
 		for result := range results {
 			analysisResults = append(analysisResults, result)
 		}
@@ -221,8 +227,10 @@ func (g *packageManagerGuard) concurrentAnalyzePackages(ctx context.Context,
 	waiter := make(chan struct{})
 	go func() {
 		wg.Wait()
-		close(waiter)
 		close(results)
+
+		rwg.Wait()
+		close(waiter)
 	}()
 
 	select {

guard/guard_test.go

@@ -0,0 +1,47 @@
+package guard
+
+import (
+	"context"
+	"testing"
+
+	packagev1 "buf.build/gen/go/safedep/api/protocolbuffers/go/safedep/messages/package/v1"
+	"github.com/safedep/pmg/analyzer"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGuardConcurrentlyAnalyzePackagesMalwareQueryService(t *testing.T) {
+	mq, err := analyzer.NewMalysisQueryAnalyzer(analyzer.MalysisQueryAnalyzerConfig{})
+	if err != nil {
+		t.Fatalf("failed to create mq: %v", err)
+	}
+
+	pg, err := NewPackageManagerGuard(DefaultPackageManagerGuardConfig(), nil, nil,
+		[]analyzer.PackageVersionAnalyzer{mq}, PackageManagerGuardInteraction{})
+	if err != nil {
+		t.Fatalf("failed to create pg: %v", err)
+	}
+
+	t.Run("should resolve a single known malicious package version", func(t *testing.T) {
+		r, err := pg.concurrentAnalyzePackages(context.Background(), []*packagev1.PackageVersion{
+			{
+				Package: &packagev1.Package{
+					Name:      "nyc-config",
+					Ecosystem: packagev1.Ecosystem_ECOSYSTEM_NPM,
+				},
+				Version: "10.0.0",
+			},
+		})
+		if err != nil {
+			t.Fatalf("failed to analyze packages: %v", err)
+		}
+
+		assert.Equal(t, 1, len(r))
+		assert.Equal(t, "nyc-config", r[0].PackageVersion.GetPackage().GetName())
+		assert.Equal(t, "10.0.0", r[0].PackageVersion.GetVersion())
+		assert.Equal(t, packagev1.Ecosystem_ECOSYSTEM_NPM, r[0].PackageVersion.GetPackage().GetEcosystem())
+		assert.NotEmpty(t, r[0].ReferenceURL)
+		assert.NotEmpty(t, r[0].Summary)
+		assert.NotNil(t, r[0].Data)
+		assert.Equal(t, analyzer.ActionBlock, r[0].Action)
+	})
+}

internal/flows/common.go

@@ -72,5 +72,10 @@ func (f *commonFlow) Run(ctx context.Context, args []string) error {
 		ui.Fatalf("Failed to create package manager guard: %s", err)
 	}
 
-	return proxy.Run(ctx, args)
+	err = proxy.Run(ctx, args)
+	if err != nil {
+		ui.Fatalf("pmg: failed to execute command: %s", err)
+	}
+
+	return err
 }

packagemanager/dependency_resolver.go

@@ -11,6 +11,10 @@ import (
 	"github.com/safedep/dry/packageregistry"
 )
 
+// Contract for a function that implements ecosystem specific version
+// resolver from a version range specification.
+type versionSpecResolver func(version string) string
+
 type dependencyResolverConfig struct {
 	IncludeDevDependencies        bool
 	IncludeTransitiveDependencies bool
@@ -23,16 +27,26 @@ type dependencyResolver struct {
 	client              packageregistry.Client
 	config              dependencyResolverConfig
 	mutex               sync.Mutex
+	versionSpecResolver versionSpecResolver
 }
 
-func newDependencyResolver(client packageregistry.Client, config dependencyResolverConfig) *dependencyResolver {
+func newDependencyResolver(client packageregistry.Client, config dependencyResolverConfig,
+	versionSpecResolver versionSpecResolver) *dependencyResolver {
 	if config.MaxConcurrency <= 0 {
 		config.MaxConcurrency = 10
 	}
 
+	if versionSpecResolver == nil {
+		// Default version spec resolver
+		versionSpecResolver = func(version string) string {
+			return version
+		}
+	}
+
 	return &dependencyResolver{
 		client:              client,
 		config:              config,
+		versionSpecResolver: versionSpecResolver,
 	}
 }
 
@@ -124,10 +138,10 @@ func (r *dependencyResolver) resolvePackageDependenciesConcurrent(
 	for _, dependency := range dependencies {
 		resolvedDependencies = append(resolvedDependencies, &packagev1.PackageVersion{
 			Package: &packagev1.Package{
-				Ecosystem: packagev1.Ecosystem_ECOSYSTEM_NPM,
+				Ecosystem: packageVersion.GetPackage().GetEcosystem(),
 				Name:      dependency.Name,
 			},
-			Version: npmCleanVersion(dependency.VersionSpec),
+			Version: r.versionSpecResolver(dependency.VersionSpec),
 		})
 	}
 

packagemanager/npm_resolver.go

@@ -78,7 +78,7 @@ func (r *npmDependencyResolver) ResolveDependencies(ctx context.Context,
 		TransitiveDepth:               r.config.TransitiveDepth,
 		FailFast:                      r.config.FailFast,
 		MaxConcurrency:                r.config.MaxConcurrency,
-	})
+	}, npmCleanVersion)
 
 	return resolver.resolveDependencies(ctx, packageVersion)
 }

test/e2e.sh

@@ -8,9 +8,9 @@ pmg=$scriptDir/../bin/pmg
 echo "Running e2e tests..."
 
 ## All these should be successful
-$pmg npm install express --dry-run
+$pmg --debug --dry-run npm install express
-$pmg --dry-run npm install express
+$pmg --debug --dry-run npm install express
-$pmg --dry-run pnpm add express
+$pmg --debug --dry-run pnpm add express
 
 ## All these should fail
-$pmg --dry-run npm install nyc-config@10.0.0
+$pmg --debug --dry-run npm install nyc-config@10.0.0 || exit 1