mirror of
https://github.com/safedep/pmg.git
synced 2026-08-03 07:24:09 +02:00
c3f3920d2e
* fix(proxy): harden MITM proxy reliability and scale for bulk installs Deep-dive analysis of dropped connections during large installs (5000+ packages with concurrent downloads) surfaced three issues, each verified with a reproduction test before fixing. 1. Transient upstream errors tore down whole keep-alive tunnels. goproxy returns false (closing the entire MITM client tunnel) when a single upstream round-trip errors. Under load, CDN-fronted registries (e.g. Cloudflare for registry.npmjs.org) intermittently reset connections, so one transient reset dropped a pooled keep-alive socket and surfaced to the package manager as ECONNRESET / "socket hang up". Fix: route upstream round-trips through a resilient round tripper that retries idempotent, body-less requests with bounded linear backoff, absorbing transient resets and keeping the tunnel alive. A reproduction test shows the tunnel count drop from 3 to 1 across a transient failure. 2. Head-of-line amplification against the external analysis service. Concurrent requests for the same package version each issued their own gRPC call. Fix: de-duplicate in-flight analyses with singleflight so a burst of identical requests collapses into one upstream call. 3. Per-request goproxy verbose logging on the hot path. proxy.Verbose was always on, formatting several log lines per request even when discarded below debug level. Fix: enable goproxy verbose logging only when PMG runs at debug level. Note: the hypothesis that the http.Server Read/WriteTimeout leaks onto hijacked CONNECT tunnels was investigated and disproven (Go clears the deadlines on Hijack); the behavioral guard tests for long-lived connections and slow transfers are retained. * fix: Type assertion error handling --------- Co-authored-by: Claude <noreply@anthropic.com>
68 lines
2.3 KiB
Go
68 lines
2.3 KiB
Go
package interceptors
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
packagev1 "buf.build/gen/go/safedep/api/protocolbuffers/go/safedep/messages/package/v1"
|
|
"github.com/safedep/pmg/analyzer"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
// slowAtomicAnalyzer counts calls atomically and blocks briefly so concurrent
|
|
// callers overlap inside analyzePackage.
|
|
type slowAtomicAnalyzer struct {
|
|
calls atomic.Int64
|
|
delay time.Duration
|
|
}
|
|
|
|
func (s *slowAtomicAnalyzer) Name() string { return "slow-atomic" }
|
|
|
|
func (s *slowAtomicAnalyzer) Analyze(_ context.Context, pv *packagev1.PackageVersion) (*analyzer.PackageVersionAnalysisResult, error) {
|
|
s.calls.Add(1)
|
|
time.Sleep(s.delay)
|
|
return &analyzer.PackageVersionAnalysisResult{PackageVersion: pv, Action: analyzer.ActionAllow}, nil
|
|
}
|
|
|
|
// TestAnalyzePackageDeduplicatesConcurrentCalls verifies that a burst of
|
|
// concurrent analyses for the same package version collapses into a single
|
|
// upstream analyzer call (head-of-line mitigation), while distinct packages
|
|
// are analyzed independently.
|
|
func TestAnalyzePackageDeduplicatesConcurrentCalls(t *testing.T) {
|
|
mock := &slowAtomicAnalyzer{delay: 50 * time.Millisecond}
|
|
base := newTestBaseInterceptor(mock)
|
|
ctx := newTestRequestContext()
|
|
|
|
const concurrency = 50
|
|
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < concurrency; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
result, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "same-pkg", "1.0.0")
|
|
assert.NoError(t, err)
|
|
require.NotNil(t, result)
|
|
assert.Equal(t, analyzer.ActionAllow, result.Action)
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
|
|
assert.EqualValues(t, 1, mock.calls.Load(),
|
|
"concurrent analyses of the same package version should collapse into one upstream call")
|
|
|
|
// A subsequent call for the same package is served from cache (no new call).
|
|
_, err := base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "same-pkg", "1.0.0")
|
|
require.NoError(t, err)
|
|
assert.EqualValues(t, 1, mock.calls.Load(), "result should be cached after the first analysis")
|
|
|
|
// A different package version triggers its own analysis.
|
|
_, err = base.analyzePackage(ctx, packagev1.Ecosystem_ECOSYSTEM_NPM, "other-pkg", "2.0.0")
|
|
require.NoError(t, err)
|
|
assert.EqualValues(t, 2, mock.calls.Load(), "distinct package versions are analyzed independently")
|
|
}
|