pmg/proxy/interceptors/base_registry.go

package interceptors

import (
	"context"
	"fmt"
	"net/http"
	"time"

	packagev1 "buf.build/gen/go/safedep/api/protocolbuffers/go/safedep/messages/package/v1"
	"github.com/safedep/dry/log"
	"github.com/safedep/pmg/analyzer"
	"github.com/safedep/pmg/config"
	"github.com/safedep/pmg/internal/audit"
	"github.com/safedep/pmg/proxy"
	gobreaker "github.com/sony/gobreaker/v2"
	"golang.org/x/sync/singleflight"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// baseRegistryInterceptor provides common functionality for registry interceptors
// It contains ecosystem-agnostic methods that can be reused by specific registry implementations
type baseRegistryInterceptor struct {
	analyzer         analyzer.PackageVersionAnalyzer
	cache            AnalysisCache
	statsCollector   *AnalysisStatsCollector
	confirmationChan chan *ConfirmationRequest
	circuitBreaker   *gobreaker.CircuitBreaker[*analyzer.PackageVersionAnalysisResult]
	execContext      InterceptorContext

	// inflight collapses concurrent analyses of the same package version into a
	// single upstream call. During a large install the same transitive
	// dependency is frequently requested across several connections at once.
	// Without de-duplication each would issue its own gRPC call to the external
	// analysis service, amplifying load and latency (head-of-line blocking).
	inflight singleflight.Group
}

func newAnalyzerCircuitBreaker(name string) *gobreaker.CircuitBreaker[*analyzer.PackageVersionAnalysisResult] {
	return newAnalyzerCircuitBreakerWithTimeout(name, 30*time.Second)
}

func newAnalyzerCircuitBreakerWithTimeout(name string, cooldown time.Duration) *gobreaker.CircuitBreaker[*analyzer.PackageVersionAnalysisResult] {
	return gobreaker.NewCircuitBreaker[*analyzer.PackageVersionAnalysisResult](gobreaker.Settings{
		Name:        name,
		MaxRequests: 1,
		Timeout:     cooldown,
		ReadyToTrip: func(counts gobreaker.Counts) bool {
			return counts.ConsecutiveFailures >= 3
		},
		OnStateChange: func(name string, from gobreaker.State, to gobreaker.State) {
			log.Infof("Circuit breaker %s: %s -> %s", name, from, to)
		},
	})
}

var _ proxy.Interceptor = (*baseRegistryInterceptor)(nil)

// Name returns a default name - should be overridden by specific implementations
func (b *baseRegistryInterceptor) Name() string {
	return "base-registry-interceptor"
}

// ShouldIntercept returns false by default - must be overridden by specific implementations
func (b *baseRegistryInterceptor) ShouldIntercept(ctx *proxy.RequestContext) bool {
	return false
}

// HandleRequest returns allow by default - should be overridden by specific implementations
func (b *baseRegistryInterceptor) HandleRequest(ctx *proxy.RequestContext) (*proxy.InterceptorResponse, error) {
	return &proxy.InterceptorResponse{Action: proxy.ActionAllow}, nil
}

// fastAllow short-circuits the request when no security control should run for
// this concrete version: insecure-installation mode (analysis globally off) or a
// trusted package (waives every control). It returns (response, true) when it
// handled the request; (nil, false) otherwise. Insecure is checked first to
// preserve the previous ordering inside analyzePackage.
func (b *baseRegistryInterceptor) fastAllow(
	ctx *proxy.RequestContext,
	ecosystem packagev1.Ecosystem,
	name, version string,
) (*proxy.InterceptorResponse, bool) {
	pkgVersion := &packagev1.PackageVersion{
		Package: &packagev1.Package{Ecosystem: ecosystem, Name: name},
		Version: version,
	}

	if config.Get().InsecureInstallation {
		log.Debugf("[%s] Skipping insecure installation", ctx.RequestID)
		audit.LogInstallInsecureBypass(pkgVersion)
		return &proxy.InterceptorResponse{Action: proxy.ActionAllow}, true
	}

	if config.IsTrustedPackageRef(ecosystem, name, version) {
		log.Debugf("[%s] Skipping trusted package: %s/%s@%s", ctx.RequestID, ecosystem.String(), name, version)
		audit.LogInstallTrustedAllowed(pkgVersion)
		return &proxy.InterceptorResponse{Action: proxy.ActionAllow}, true
	}

	return nil, false
}

// analyzePackage analyzes a package using the configured analyzer with caching
// This method is ecosystem-agnostic and can be used by any registry interceptor
func (b *baseRegistryInterceptor) analyzePackage(
	ctx *proxy.RequestContext,
	ecosystem packagev1.Ecosystem,
	packageName string,
	packageVersion string,
) (*analyzer.PackageVersionAnalysisResult, error) {
	pkgVersion := &packagev1.PackageVersion{
		Package: &packagev1.Package{
			Ecosystem: ecosystem,
			Name:      packageName,
		},
		Version: packageVersion,
	}

	if cached, ok := b.cache.Get(ecosystem.String(), packageName, packageVersion); ok {
		log.Debugf("[%s] Using cached analysis result for %s@%s", ctx.RequestID, packageName, packageVersion)
		return cached, nil
	}

	log.Debugf("[%s] Analyzing package %s@%s", ctx.RequestID, packageName, packageVersion)

	key := ecosystem.String() + ":" + packageName + ":" + packageVersion
	resultAny, err, _ := b.inflight.Do(key, func() (interface{}, error) {
		// Re-check the cache: a previous in-flight analysis for this key may
		// have populated it after our own cache miss above.
		if cached, ok := b.cache.Get(ecosystem.String(), packageName, packageVersion); ok {
			return cached, nil
		}

		result, err := b.circuitBreaker.Execute(func() (*analyzer.PackageVersionAnalysisResult, error) {
			analysisCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			defer cancel()

			res, err := b.analyzer.Analyze(analysisCtx, pkgVersion)
			if err != nil {
				// NotFound means the package is not in the analysis DB — this is expected
				// and should not count as a circuit breaker failure.
				// Since gRPC v1.75.0, status.FromError unwraps error chains via errors.As.
				if s, ok := status.FromError(err); ok && s.Code() == codes.NotFound {
					log.Debugf("[%s] Package %s@%s not found in analysis DB, allowing", ctx.RequestID, packageName, packageVersion)
					return &analyzer.PackageVersionAnalysisResult{
						PackageVersion: pkgVersion,
						Action:         analyzer.ActionAllow,
					}, nil
				}
			}

			return res, err
		})
		if err != nil {
			return nil, err
		}

		b.cache.Set(ecosystem.String(), packageName, packageVersion, result)
		return result, nil
	})
	if err != nil {
		return nil, fmt.Errorf("analyzer failed: %w", err)
	}

	// Fail loudly rather than panic in the proxy hot path if a future change to
	// the singleflight closure ever returns a different type or a nil result.
	result, ok := resultAny.(*analyzer.PackageVersionAnalysisResult)
	if !ok || result == nil {
		return nil, fmt.Errorf("analyzer returned unexpected result type %T for %s@%s", resultAny, packageName, packageVersion)
	}

	log.Debugf("[%s] Analysis complete for %s@%s: action=%d", ctx.RequestID, packageName, packageVersion, result.Action)

	return result, nil
}

// handleAnalysisResult processes the analysis result and returns appropriate response action
// This method is ecosystem agnostic and handles the analysis result uniformly
func (b *baseRegistryInterceptor) handleAnalysisResult(
	ctx *proxy.RequestContext,
	ecosystem packagev1.Ecosystem,
	packageName string,
	packageVersion string,
	result *analyzer.PackageVersionAnalysisResult,
) (*proxy.InterceptorResponse, error) {
	switch result.Action {
	case analyzer.ActionBlock:
		log.Warnf("[%s] Blocking malicious package %s@%s", ctx.RequestID, packageName, packageVersion)

		audit.LogMalwareBlocked(result.PackageVersion, result.Summary, result.AnalysisID, result.ReferenceURL, result.IsMalware, result.IsVerified)

		if b.statsCollector != nil {
			b.statsCollector.RecordBlocked(result)
		}

		message := fmt.Sprintf("Malicious package blocked: %s/%s@%s\n\nReason: %s\n\nReference: %s",
			ecosystem.String(),
			packageName, packageVersion,
			result.Summary,
			result.ReferenceURL)

		return &proxy.InterceptorResponse{
			Action:       proxy.ActionBlock,
			BlockCode:    http.StatusForbidden,
			BlockMessage: message,
		}, nil

	case analyzer.ActionConfirm:
		log.Warnf("[%s] Package %s/%s@%s is suspicious, requesting user confirmation", ctx.RequestID, ecosystem.String(), packageName, packageVersion)

		confirmed, err := b.requestUserConfirmation(ctx, result)
		if err != nil {
			log.Errorf("[%s] Failed to get user confirmation: %v", ctx.RequestID, err)

			if b.statsCollector != nil {
				b.statsCollector.RecordBlocked(result)
			}

			return &proxy.InterceptorResponse{
				Action:       proxy.ActionBlock,
				BlockCode:    http.StatusForbidden,
				BlockMessage: fmt.Sprintf("Failed to get user confirmation for suspicious package %s/%s@%s", ecosystem.String(), packageName, packageVersion),
			}, nil
		}

		if !confirmed {
			log.Infof("[%s] User declined installation of suspicious package %s/%s@%s", ctx.RequestID, ecosystem.String(), packageName, packageVersion)

			audit.LogMalwareBlocked(result.PackageVersion, result.Summary, result.AnalysisID, result.ReferenceURL, result.IsMalware, result.IsVerified)

			if b.statsCollector != nil {
				b.statsCollector.RecordUserCancelled(result)
			}

			message := fmt.Sprintf("Installation blocked by user: %s/%s@%s\n\nReason: %s\n\nReference: %s",
				ecosystem.String(),
				packageName, packageVersion,
				result.Summary,
				result.ReferenceURL)

			return &proxy.InterceptorResponse{
				Action:       proxy.ActionBlock,
				BlockCode:    http.StatusForbidden,
				BlockMessage: message,
			}, nil
		}

		audit.LogMalwareConfirmed(result.PackageVersion, result.AnalysisID, result.IsMalware, result.IsVerified)
		audit.LogInstallAllowed(result.PackageVersion, 1)

		if b.statsCollector != nil {
			b.statsCollector.RecordConfirmed(result)
		}

		log.Infof("[%s] User confirmed installation of suspicious package %s/%s@%s", ctx.RequestID, ecosystem.String(), packageName, packageVersion)
		return &proxy.InterceptorResponse{Action: proxy.ActionAllow}, nil

	case analyzer.ActionAllow:
		audit.LogInstallAllowed(result.PackageVersion, 1)

		if b.statsCollector != nil {
			b.statsCollector.RecordAllowed(result)
		}

		// A flagged package allowed only because of a tenant-specific exclusion is
		// a security-relevant trust decision; surface it so it is never silent.
		if result.IsExcluded {
			log.Warnf("[%s] Allowing flagged package %s/%s@%s due to tenant exclusion (%s)",
				ctx.RequestID, ecosystem.String(), packageName, packageVersion, result.ExclusionReason)
		} else {
			log.Debugf("[%s] Package %s/%s@%s is safe, allowing request", ctx.RequestID, ecosystem.String(), packageName, packageVersion)
		}
		return &proxy.InterceptorResponse{Action: proxy.ActionAllow}, nil

	default:
		audit.LogInstallAllowed(result.PackageVersion, 1)

		if b.statsCollector != nil {
			b.statsCollector.RecordAllowed(result)
		}

		log.Warnf("[%s] Unknown analysis action %d for package %s/%s@%s, allowing by default", ctx.RequestID, result.Action, ecosystem.String(), packageName, packageVersion)
		return &proxy.InterceptorResponse{Action: proxy.ActionAllow}, nil
	}
}

// requestUserConfirmation sends a confirmation request and blocks waiting for user response
func (b *baseRegistryInterceptor) requestUserConfirmation(
	ctx *proxy.RequestContext,
	result *analyzer.PackageVersionAnalysisResult,
) (bool, error) {
	req := NewConfirmationRequest(result.PackageVersion, result)

	select {
	case b.confirmationChan <- req:
	case <-time.After(5 * time.Second):
		return false, fmt.Errorf("timeout sending confirmation request")
	}

	// Block waiting for user response
	// Producer is responsible for closing the response channel to prevent goroutine leaks.
	select {
	case confirmed := <-req.ResponseChan:
		return confirmed, nil
	case <-time.After(5 * time.Minute):
		return false, fmt.Errorf("timeout waiting for user confirmation")
	}
}