fix: keep http.DefaultTransport a *http.Transport for OPA http.send

README.md

@@ -372,7 +372,7 @@ All carry `strategy` and `registry`. Registry values come from configured host/A
 
 The HTTP server metrics of these handlers carry `cachew.cache.result` (`hit`, `miss`, `coalesced` or `error`). Hits also carry `cachew.cache.source`, the backend that served them (`disk`, `s3`, `memory`, `remote`, or `other`), after stale and conditional probes.
 
-Origin metrics count HTTP exchanges that reach an origin through `http.DefaultTransport`, including redirect hops and transport failures. Requests rejected during an upstream 429 cooldown are not counted. `status_code=none` means no HTTP response; `error_type` is `none`, `transport`, `timeout`, or `canceled`. Duration ends at response headers or transport failure. Registry identifies the logical artifact provider, including its redirects, rather than each physical destination. These metrics do not count every low-level connection retry, Git transport, Go-module fetch, GitHub metadata lookup, OCI tag-list or referrer pass-through, or S3 operation. OCI registry ping and token exchanges are counted, because they go to the registry for the artifact. Internal Hermit forwarding to another Cachew route is excluded from origin/fill counts; the receiving handler measures its own work.
+Origin metrics count HTTP exchanges that reach an origin through the shared upstream transport, including redirect hops and transport failures. Requests rejected during an upstream 429 cooldown are not counted. `status_code=none` means no HTTP response; `error_type` is `none`, `transport`, `timeout`, or `canceled`. Duration ends at response headers or transport failure. Registry identifies the logical artifact provider, including its redirects, rather than each physical destination. These metrics do not count every low-level connection retry, Git transport, Go-module fetch, GitHub metadata lookup, OCI tag-list or referrer pass-through, or S3 operation. OCI registry ping and token exchanges are counted, because they go to the registry for the artifact. Internal Hermit forwarding to another Cachew route is excluded from origin/fill counts; the receiving handler measures its own work.
 
 Fills record a terminal outcome, rather than treating an origin 200 as successful publication. `purpose` separates `origin`, `backfill` and ranged `heal`. Results distinguish successful completion from creation, write, commit, transfer or read errors, cancellation, and superseded heals. Tier writes report backend outcomes separately: a failed S3 write can coexist with a successful disk write. Early close or canceled backfill is not a successful fill. Client responses continue normally when cache writes fail.
 

client/client.go

@@ -65,8 +65,8 @@ type CacheWriter interface {
 	Abort(err error) error
 }
 
-// stdTransport is captured before cachewd wraps http.DefaultTransport with
-// upstream 429 cooldowns, so cache clients keep their own tuned pool.
+// stdTransport is the process default transport, kept so cache clients have
+// their own tuned pool even if a test has swapped http.DefaultTransport.
 var stdTransport = http.DefaultTransport.(*http.Transport) //nolint:errcheck,gochecknoglobals
 
 // HeaderFunc returns headers to attach to each outgoing request.

cmd/cachewd/main.go

@@ -460,10 +460,13 @@ func configureProcess(logger *slog.Logger, mutexProfileFraction int, upstreamBac
 	// side-effect import; OPA gates external access.
 	runtime.SetMutexProfileFraction(mutexProfileFraction)
 
-	// Every strategy that uses http.DefaultTransport shares one set of
+	// Every strategy that uses httputil.UpstreamTransport shares one set of
 	// per-hostname and per-credential 429 cooldowns. Origin metrics sit below the cooldowns so that
 	// requests rejected during a cooldown are not counted as origin traffic.
-	http.DefaultTransport = httputil.NewBackoffTransport(metrics.NewOriginTransport(http.DefaultTransport), upstreamBackoff, upstreamBackoffMax, logger) //nolint:reassign
+	// http.DefaultTransport itself stays a *http.Transport: OPA's http.send
+	// asserts that type whenever a policy sets a TLS option, and panics on a
+	// wrapper.
+	httputil.SetUpstreamTransport(httputil.NewBackoffTransport(metrics.NewOriginTransport(http.DefaultTransport), upstreamBackoff, upstreamBackoffMax, logger))
 }
 
 // loadGlobalConfig unmarshals the global config from HCL, using a two-pass

cmd/cachewd/opa_transport_test.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"encoding/pem"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/alecthomas/assert/v2"
+	"github.com/open-policy-agent/opa/v1/rego"
+
+	"github.com/block/cachew/internal/httputil"
+	"github.com/block/cachew/internal/logging"
+)
+
+const opaTransportStartupChildEnv = "CACHEWD_OPA_TRANSPORT_STARTUP_CHILD"
+
+func TestOPAHTTPSendWithTLSOptionsAfterProcessSetup(t *testing.T) {
+	if os.Getenv(opaTransportStartupChildEnv) != "1" {
+		cmd := exec.CommandContext(t.Context(), os.Args[0], "-test.run=^"+t.Name()+"$", "-test.count=1")
+		cmd.Env = append(os.Environ(), opaTransportStartupChildEnv+"=1")
+		output, err := cmd.CombinedOutput()
+		assert.NoError(t, err, "%s", output)
+		return
+	}
+	logger, ctx := logging.Configure(t.Context(), logging.Config{Level: slog.LevelError})
+	configureProcess(logger, 0, time.Second, time.Minute)
+
+	_, ok := http.DefaultTransport.(*http.Transport)
+	assert.True(t, ok, "configureProcess must leave http.DefaultTransport a *http.Transport, got %T", http.DefaultTransport)
+	_, ok = httputil.UpstreamTransport().(*http.Transport)
+	assert.False(t, ok, "configureProcess must install the upstream transport")
+
+	var upstreamHits int
+	upstream := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		upstreamHits++
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Cleanup(upstream.Close)
+	caFile := filepath.Join(t.TempDir(), "ca.crt")
+	assert.NoError(t, os.WriteFile(caFile, pem.EncodeToMemory(&pem.Block{
+		Type: "CERTIFICATE", Bytes: upstream.Certificate().Raw,
+	}), 0o600))
+
+	query, err := rego.New(
+		rego.Query("data.cachew.allow"),
+		rego.Module("policy.rego", `
+package cachew
+
+default allow := false
+
+allow if {
+	resp := http.send({"method": "HEAD", "url": input.url, "tls_ca_cert_file": input.ca_file})
+	resp.status_code == 200
+}
+`),
+	).PrepareForEval(ctx)
+	assert.NoError(t, err)
+
+	results, err := query.Eval(ctx, rego.EvalInput(map[string]any{"url": upstream.URL, "ca_file": caFile}))
+	assert.NoError(t, err)
+	assert.Equal(t, 1, len(results))
+	assert.Equal(t, true, results[0].Expressions[0].Value)
+	assert.Equal(t, 1, upstreamHits)
+}

internal/githubapp/tokens.go

@@ -24,7 +24,7 @@ type TokenManagerProvider func() (*TokenManager, error)
 // from one or more GitHub App configurations.
 func NewTokenManagerProvider(configs []Config, logger *slog.Logger) TokenManagerProvider {
 	return sync.OnceValues(func() (*TokenManager, error) {
-		return newTokenManager(configs, logger, githubAPIBase, http.DefaultClient)
+		return newTokenManager(configs, logger, githubAPIBase, httputil.UpstreamClient())
 	})
 }
 

internal/githubapp/transport.go

@@ -33,7 +33,7 @@ func NewAPIClient(base *http.Client) *http.Client {
 	client := *base
 	transport := client.Transport
 	if transport == nil {
-		transport = http.DefaultTransport
+		transport = httputil.UpstreamTransport()
 	}
 	client.Transport = &apiTransport{
 		base: transport,

internal/httputil/upstream.go

@@ -0,0 +1,45 @@
+package httputil
+
+import (
+	"net/http"
+	"sync/atomic"
+)
+
+// upstream is the transport cachewd installs for outbound requests that would
+// otherwise use http.DefaultTransport. It is kept here instead of being
+// assigned to http.DefaultTransport because third-party code, such as OPA's
+// http.send, asserts that the global is a *http.Transport and panics on any
+// wrapper.
+var upstream atomic.Pointer[http.RoundTripper] //nolint:gochecknoglobals
+
+// SetUpstreamTransport installs the shared outbound transport returned by
+// [UpstreamTransport], leaving http.DefaultTransport unchanged. A nil
+// transport restores the default of following http.DefaultTransport.
+func SetUpstreamTransport(rt http.RoundTripper) {
+	if rt == nil {
+		upstream.Store(nil)
+		return
+	}
+	upstream.Store(&rt)
+}
+
+// UpstreamTransport returns the shared outbound transport. Until one is
+// installed, requests go to whatever http.DefaultTransport is at request time,
+// so tests can still swap the global to intercept upstream traffic.
+func UpstreamTransport() http.RoundTripper {
+	return upstreamTransport{}
+}
+
+// UpstreamClient returns a client that sends through [UpstreamTransport].
+func UpstreamClient() *http.Client {
+	return &http.Client{Transport: UpstreamTransport()}
+}
+
+type upstreamTransport struct{}
+
+func (upstreamTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	if rt := upstream.Load(); rt != nil {
+		return (*rt).RoundTrip(req) //nolint:wrapcheck
+	}
+	return http.DefaultTransport.RoundTrip(req) //nolint:wrapcheck
+}

internal/httputil/upstream_test.go

@@ -0,0 +1,38 @@
+package httputil_test
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/alecthomas/assert/v2"
+
+	"github.com/block/cachew/internal/httputil"
+)
+
+type staticTransport struct{ status int }
+
+func (s staticTransport) RoundTrip(*http.Request) (*http.Response, error) {
+	return &http.Response{StatusCode: s.status, Body: http.NoBody}, nil
+}
+
+func TestUpstreamTransportFollowsDefaultUntilSet(t *testing.T) {
+	defaultTransport := http.DefaultTransport
+	t.Cleanup(func() {
+		http.DefaultTransport = defaultTransport //nolint:reassign
+		httputil.SetUpstreamTransport(nil)
+	})
+	httputil.SetUpstreamTransport(nil)
+	http.DefaultTransport = staticTransport{status: http.StatusTeapot} //nolint:reassign
+	req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://example.com/", nil)
+	assert.NoError(t, err)
+
+	resp, err := httputil.UpstreamClient().Do(req)
+	assert.NoError(t, err)
+	assert.Equal(t, http.StatusTeapot, resp.StatusCode)
+
+	httputil.SetUpstreamTransport(staticTransport{status: http.StatusAccepted})
+	resp, err = httputil.UpstreamClient().Do(req)
+	assert.NoError(t, err)
+	assert.Equal(t, http.StatusAccepted, resp.StatusCode)
+	assert.Equal[http.RoundTripper](t, staticTransport{status: http.StatusTeapot}, http.DefaultTransport)
+}

internal/strategy/artifactory.go

@@ -70,7 +70,7 @@ func NewArtifactory(ctx context.Context, config ArtifactoryConfig, cache cache.C
 	a := &Artifactory{
 		target: u,
 		cache:  cache,
-		client: &http.Client{},
+		client: httputil.UpstreamClient(),
 		logger: logging.FromContext(ctx),
 	}
 

internal/strategy/git/git.go

@@ -160,7 +160,7 @@ func New(
 		config:         config,
 		cache:          cache,
 		cloneManager:   cloneManager,
-		httpClient:     http.DefaultClient,
+		httpClient:     cachewhttputil.UpstreamClient(),
 		ctx:            ctx,
 		scheduler:      scheduler.WithQueuePrefix("git"),
 		spoolRegistry:  spools,

internal/strategy/github_releases.go

@@ -61,7 +61,7 @@ func NewGitHubReleases(ctx context.Context, config GitHubReleasesConfig, cache c
 	s := &GitHubReleases{
 		config:       config,
 		cache:        cache,
-		client:       githubapp.NewAPIClient(http.DefaultClient),
+		client:       githubapp.NewAPIClient(httputil.UpstreamClient()),
 		tokenManager: tokenManager,
 	}
 	// eg. https://github.com/alecthomas/chroma/releases/download/v2.21.1/chroma-2.21.1-darwin-amd64.tar.gz

internal/strategy/gomod/gomod.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/block/cachew/internal/cache"
 	"github.com/block/cachew/internal/gitclone"
+	"github.com/block/cachew/internal/httputil"
 	"github.com/block/cachew/internal/logging"
 	"github.com/block/cachew/internal/strategy"
 )
@@ -64,6 +65,7 @@ func New(ctx context.Context, config Config, cache cache.Cache, mux strategy.Mux
 	}
 
 	publicFetcher := &goproxy.GoFetcher{
+		Transport: httputil.UpstreamTransport(),
 		Env: []string{
 			"GOPROXY=" + config.Proxy,
 			"GOSUMDB=off", // Disable checksum database validation in fetcher, to prevent unneccessary double validation
@@ -81,8 +83,9 @@ func New(ctx context.Context, config Config, cache cache.Cache, mux strategy.Mux
 	}
 
 	s.goproxy = &goproxy.Goproxy{
-		Logger:  s.logger,
-		Fetcher: fetcher,
+		Logger:    s.logger,
+		Fetcher:   fetcher,
+		Transport: httputil.UpstreamTransport(),
 		Cacher: &goproxyCacher{
 			cache: cache,
 		},

internal/strategy/hermit.go

@@ -62,7 +62,7 @@ func NewHermit(ctx context.Context, config HermitConfig, _ jobscheduler.Schedule
 	s := &Hermit{
 		config: config,
 		cache:  c,
-		client: http.DefaultClient,
+		client: httputil.UpstreamClient(),
 		logger: logger,
 		mux:    mux,
 	}

internal/strategy/host.go

@@ -69,13 +69,13 @@ func NewHost(_ context.Context, config HostConfig, cache cache.Cache, mux Mux, s
 	h := &Host{
 		target:  u,
 		cache:   cache,
-		client:  &http.Client{},
+		client:  httputil.UpstreamClient(),
 		prefix:  prefix,
 		headers: config.Headers,
 	}
 
 	if len(config.Fallbacks) > 0 {
-		transport := &hostFallbackTransport{client: &http.Client{Transport: http.DefaultTransport}}
+		transport := &hostFallbackTransport{client: httputil.UpstreamClient()}
 		for _, config := range config.Fallbacks {
 			target, err := url.Parse(config.Target)
 			if err != nil || target.Host == "" || (target.Scheme != "http" && target.Scheme != "https") ||

internal/strategy/oci.go

@@ -103,7 +103,7 @@ func NewOCI(ctx context.Context, config OCIConfig, c cache.Cache, mux Mux, spool
 	if err != nil {
 		return nil, err
 	}
-	transport, err := newOCITransport(upstream, config.Username, config.Password, http.DefaultTransport)
+	transport, err := newOCITransport(upstream, config.Username, config.Password, httputil.UpstreamTransport())
 	if err != nil {
 		return nil, err
 	}

internal/strategy/proxy.go

@@ -10,6 +10,7 @@ import (
 	"github.com/alecthomas/errors"
 
 	"github.com/block/cachew/internal/cache"
+	"github.com/block/cachew/internal/httputil"
 	"github.com/block/cachew/internal/logging"
 	"github.com/block/cachew/internal/spool"
 	"github.com/block/cachew/internal/strategy/handler"
@@ -56,7 +57,7 @@ var (
 
 func NewHTTPProxy(ctx context.Context, c cache.Cache, _ Mux, spools *spool.Registry) (*HTTPProxy, error) {
 	logger := logging.FromContext(ctx)
-	client := &http.Client{}
+	client := httputil.UpstreamClient()
 	p := &HTTPProxy{logger: logger}
 
 	p.handler = handler.New(client, c, spools).Metrics("proxy", "other").