From c103b38c95924dc56f548811eef14af686046218 Mon Sep 17 00:00:00 2001 From: dawn Date: Wed, 1 Jul 2026 16:34:47 +0300 Subject: [PATCH] spindle/microvm: allow configuring headers for read/upload cache reqs Signed-off-by: dawn --- cmd/spindle-microvm-run/main_linux.go | 4 +- docs/DOCS.md | 14 +++++ spindle/config/config.go | 12 ++-- spindle/engines/microvm/engine.go | 6 +- spindle/engines/microvm/read_cache_proxy.go | 14 +++-- .../engines/microvm/read_cache_proxy_test.go | 56 +++++++++++++++++-- spindle/engines/microvm/upload_cache_http.go | 9 ++- .../microvm/upload_cache_nix_store_test.go | 4 +- spindle/engines/microvm/upload_cache_proxy.go | 8 +-- .../microvm/upload_cache_proxy_test.go | 32 ++++++++++- 10 files changed, 126 insertions(+), 33 deletions(-) diff --git a/cmd/spindle-microvm-run/main_linux.go b/cmd/spindle-microvm-run/main_linux.go index bcda9244..795b858c 100644 --- a/cmd/spindle-microvm-run/main_linux.go +++ b/cmd/spindle-microvm-run/main_linux.go @@ -190,7 +190,7 @@ func runMicroVMRunDev(ctx context.Context, cmd *cli.Command) error { } defer conn.Close() - upstreams, err := microvm.BuildCacheUpstreams(cmd.StringSlice("cache-read-url"), nil) + upstreams, err := microvm.BuildCacheUpstreams(cmd.StringSlice("cache-read-url"), nil, nil) if err != nil { return fmt.Errorf("build cache upstreams: %w", err) } @@ -208,7 +208,7 @@ func runMicroVMRunDev(ctx context.Context, cmd *cli.Command) error { var uploadCache *microvm.UploadCacheProxy if cmd.String("cache-upload-url") != "" { var err error - uploadCache, err = microvm.StartUploadCacheProxy(ctx, vm.CID(), cmd.String("cache-upload-url"), upstreams, filepath.Join(vm.WorkDir(), "upload-cache"), logger) + uploadCache, err = microvm.StartUploadCacheProxy(ctx, vm.CID(), cmd.String("cache-upload-url"), upstreams, filepath.Join(vm.WorkDir(), "upload-cache"), nil, logger) if err != nil { return fmt.Errorf("start upload cache proxy: %w", err) } diff --git a/docs/DOCS.md b/docs/DOCS.md index 2bf4c2bc..adbf83ba 100644 --- a/docs/DOCS.md +++ b/docs/DOCS.md @@ -1501,6 +1501,20 @@ cache (and read from it), configure the cache (prefix trusted public keys for those caches. - `SPINDLE_NIX_CACHE_UPLOAD_URL`: Cache URL that paths built in the guest are uploaded to. +- `SPINDLE_NIX_CACHE_READ_REQUEST_HEADERS`: Extra HTTP headers + sent with every read request to the operator-configured read + caches. Format is comma-separated `Name:Value` pairs, split + on the first colon. Headers set here are **not** forwarded + to workflow-supplied caches (e.g. those declared under + `caches:` in a workflow file). Useful for caches that might + require auth (eg. [attic](https://github.com/zhaofengli/attic)) + Example: + ``` + SPINDLE_NIX_CACHE_READ_REQUEST_HEADERS="Authorization:Bearer mytoken" + ``` +- `SPINDLE_NIX_CACHE_UPLOAD_REQUEST_HEADERS`: Extra HTTP headers + sent with every upload request to the upload cache. Same + format as above. ### Running spindle diff --git a/spindle/config/config.go b/spindle/config/config.go index 1d8ef04d..f1b08a60 100644 --- a/spindle/config/config.go +++ b/spindle/config/config.go @@ -87,9 +87,11 @@ type MicroVMPipelines struct { } type NixCache struct { - ReadURLs []string `env:"READ_URLS"` - TrustedPublicKeys []string `env:"TRUSTED_PUBLIC_KEYS"` - UploadURL string `env:"UPLOAD_URL"` + ReadURLs []string `env:"READ_URLS"` + TrustedPublicKeys []string `env:"TRUSTED_PUBLIC_KEYS"` + UploadURL string `env:"UPLOAD_URL"` + UploadRequestHeaders map[string]string `env:"UPLOAD_REQUEST_HEADERS"` + ReadRequestHeaders map[string]string `env:"READ_REQUEST_HEADERS"` } type Config struct { @@ -102,10 +104,8 @@ type Config struct { func Load(ctx context.Context) (*Config, error) { var cfg Config - err := envconfig.Process(ctx, &cfg) - if err != nil { + if err := envconfig.Process(ctx, &cfg); err != nil { return nil, err } - return &cfg, nil } diff --git a/spindle/engines/microvm/engine.go b/spindle/engines/microvm/engine.go index 434ccd13..adf8f1d4 100644 --- a/spindle/engines/microvm/engine.go +++ b/spindle/engines/microvm/engine.go @@ -232,7 +232,7 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m } }() - upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs) + upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs, e.cfg.NixCache.ReadRequestHeaders) if err != nil { return err } @@ -242,7 +242,7 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m } state.ReadCache = readCache stagingDir := filepath.Join(workDir, "upload-cache") - uploadCache, err := StartUploadCacheProxy(ctx, cid, e.cfg.NixCache.UploadURL, upstreams, stagingDir, l) + uploadCache, err := StartUploadCacheProxy(ctx, cid, e.cfg.NixCache.UploadURL, upstreams, stagingDir, e.cfg.NixCache.UploadRequestHeaders, l) if err != nil { return err } @@ -492,7 +492,7 @@ func (e *Engine) activateConfig(ctx context.Context, wid models.WorkflowId, stat } func (e *Engine) anyCacheHasPath(ctx context.Context, state *workflowState, storePath string) bool { - upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs) + upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs, e.cfg.NixCache.ReadRequestHeaders) if err != nil { e.l.Warn("config cache check: build upstreams failed; treating as absent", "path", storePath, "error", err) return false diff --git a/spindle/engines/microvm/read_cache_proxy.go b/spindle/engines/microvm/read_cache_proxy.go index 2d1d749f..eefa3bae 100644 --- a/spindle/engines/microvm/read_cache_proxy.go +++ b/spindle/engines/microvm/read_cache_proxy.go @@ -153,10 +153,11 @@ type CacheUpstream struct { url *url.URL // guarded upstreams come from the workflow file // requests to them are refused for special-purpose address ranges - guarded bool + guarded bool + extraHeaders map[string]string } -func BuildCacheUpstreams(rawTrusted, rawGuarded []string) ([]CacheUpstream, error) { +func BuildCacheUpstreams(rawTrusted, rawGuarded []string, extraHeaders map[string]string) ([]CacheUpstream, error) { trusted, err := parseCacheUpstreams(rawTrusted) if err != nil { return nil, err @@ -165,10 +166,10 @@ func BuildCacheUpstreams(rawTrusted, rawGuarded []string) ([]CacheUpstream, erro if err != nil { return nil, err } - return mergeCacheUpstreams(trusted, guarded), nil + return mergeCacheUpstreams(trusted, guarded, extraHeaders), nil } -func mergeCacheUpstreams(trusted, guarded []*url.URL) []CacheUpstream { +func mergeCacheUpstreams(trusted, guarded []*url.URL, trustedHeaders map[string]string) []CacheUpstream { merged := make([]CacheUpstream, 0, len(trusted)+len(guarded)) seen := make(map[string]struct{}, len(trusted)+len(guarded)) for _, u := range trusted { @@ -176,7 +177,7 @@ func mergeCacheUpstreams(trusted, guarded []*url.URL) []CacheUpstream { continue } seen[u.String()] = struct{}{} - merged = append(merged, CacheUpstream{url: u}) + merged = append(merged, CacheUpstream{url: u, extraHeaders: trustedHeaders}) } for _, u := range guarded { if _, ok := seen[u.String()]; ok { @@ -352,6 +353,9 @@ func (t *parallelRacingTransport) RoundTrip(req *http.Request) (*http.Response, password, _ := user.Password() raceReq.SetBasicAuth(user.Username(), password) } + for name, value := range target.extraHeaders { + raceReq.Header.Set(name, value) + } rt := t.underlying if target.guarded { diff --git a/spindle/engines/microvm/read_cache_proxy_test.go b/spindle/engines/microvm/read_cache_proxy_test.go index 9b92b137..f0592064 100644 --- a/spindle/engines/microvm/read_cache_proxy_test.go +++ b/spindle/engines/microvm/read_cache_proxy_test.go @@ -28,7 +28,7 @@ func TestCacheProxyFallsBackOnNotFound(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil) rec := httptest.NewRecorder() - cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) + cacheProxyHandler(mergeCacheUpstreams(upstreams, nil, nil), slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) @@ -51,7 +51,7 @@ func TestCacheProxyServesNixCacheInfoItself(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "http://guest/nix-cache-info", nil) rec := httptest.NewRecorder() - cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) + cacheProxyHandler(mergeCacheUpstreams(upstreams, nil, nil), slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) @@ -79,7 +79,7 @@ func TestCacheProxyErrorStatusDoesNotWinRace(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil) rec := httptest.NewRecorder() - cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) + cacheProxyHandler(mergeCacheUpstreams(upstreams, nil, nil), slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) @@ -112,7 +112,7 @@ func TestCacheProxyJoinsSubpathQueryAndAuth(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil) rec := httptest.NewRecorder() - cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) + cacheProxyHandler(mergeCacheUpstreams(upstreams, nil, nil), slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) @@ -137,7 +137,7 @@ func TestCacheProxyGuardedUpstreamCannotReachBlockedRanges(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil) rec := httptest.NewRecorder() - cacheProxyHandler(mergeCacheUpstreams(nil, upstreams), slog.Default()).ServeHTTP(rec, req) + cacheProxyHandler(mergeCacheUpstreams(nil, upstreams, nil), slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusBadGateway { t.Fatalf("status: got %d, want 502; body=%q", rec.Code, rec.Body.String()) @@ -163,9 +163,53 @@ func TestCacheProxyRewritesHostHeader(t *testing.T) { req := httptest.NewRequest(http.MethodGet, "http://127.0.0.1:10500/abc.narinfo", nil) req.Host = "127.0.0.1:10500" rec := httptest.NewRecorder() - cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req) + cacheProxyHandler(mergeCacheUpstreams(upstreams, nil, nil), slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) } } + +func TestReadCacheProxyInjectsExtraHeadersForTrustedOnly(t *testing.T) { + var trustedGotAuth, guardedGotAuth string + + trustedSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + trustedGotAuth = req.Header.Get("Authorization") + _, _ = io.WriteString(w, "from-trusted") + })) + defer trustedSrv.Close() + + // verify guarded upstreams never receive the injected headers + guardedSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + guardedGotAuth = req.Header.Get("Authorization") + // return 404 so trusted wins the race + w.WriteHeader(http.StatusNotFound) + })) + defer guardedSrv.Close() + + trustedURLs, err := parseCacheUpstreams([]string{trustedSrv.URL}) + if err != nil { + t.Fatal(err) + } + guardedURLs, err := parseCacheUpstreams([]string{guardedSrv.URL}) + if err != nil { + t.Fatal(err) + } + + extraHeaders := map[string]string{"Authorization": "Bearer readtoken"} + handler := cacheProxyHandler(mergeCacheUpstreams(trustedURLs, guardedURLs, extraHeaders), slog.Default()) + + req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil) + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) + } + if trustedGotAuth != "Bearer readtoken" { + t.Fatalf("trusted Authorization: got %q, want %q", trustedGotAuth, "Bearer readtoken") + } + if guardedGotAuth != "" { + t.Fatalf("guarded Authorization should be empty, got %q", guardedGotAuth) + } +} diff --git a/spindle/engines/microvm/upload_cache_http.go b/spindle/engines/microvm/upload_cache_http.go index 0310ef09..caf1b929 100644 --- a/spindle/engines/microvm/upload_cache_http.go +++ b/spindle/engines/microvm/upload_cache_http.go @@ -17,8 +17,8 @@ type httpUploadBackend struct { handler http.Handler } -func newHTTPUploadProxyBackend(target *url.URL, readUpstreams []CacheUpstream, logger *slog.Logger) *httpUploadBackend { - return &httpUploadBackend{handler: uploadProxyHandler(target, readUpstreams, logger)} +func newHTTPUploadProxyBackend(target *url.URL, readUpstreams []CacheUpstream, extraHeaders map[string]string, logger *slog.Logger) *httpUploadBackend { + return &httpUploadBackend{handler: uploadProxyHandler(target, readUpstreams, extraHeaders, logger)} } func (b *httpUploadBackend) ServeHTTP(w http.ResponseWriter, r *http.Request) { @@ -27,7 +27,7 @@ func (b *httpUploadBackend) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (b *httpUploadBackend) Close() error { return nil } -func uploadProxyHandler(target *url.URL, readUpstreams []CacheUpstream, logger *slog.Logger) http.Handler { +func uploadProxyHandler(target *url.URL, readUpstreams []CacheUpstream, extraHeaders map[string]string, logger *slog.Logger) http.Handler { rp := httputil.NewSingleHostReverseProxy(target) rp.ErrorLog = slog.NewLogLogger(logger.Handler(), slog.LevelError) @@ -42,6 +42,9 @@ func uploadProxyHandler(target *url.URL, readUpstreams []CacheUpstream, logger * password, _ := user.Password() req.SetBasicAuth(user.Username(), password) } + for name, value := range extraHeaders { + req.Header.Set(name, value) + } } // before uploading, nix copy asks the destination whether it already has each diff --git a/spindle/engines/microvm/upload_cache_nix_store_test.go b/spindle/engines/microvm/upload_cache_nix_store_test.go index 413fb66d..dbd6f207 100644 --- a/spindle/engines/microvm/upload_cache_nix_store_test.go +++ b/spindle/engines/microvm/upload_cache_nix_store_test.go @@ -44,7 +44,7 @@ func TestUploadCacheBackendSchemeDispatch(t *testing.T) { for _, tc := range cases { t.Run(tc.uploadURL, func(t *testing.T) { - backend, err := newUploadCacheBackend(tc.uploadURL, nil, staging, logger) + backend, err := newUploadCacheBackend(tc.uploadURL, nil, staging, nil, logger) if tc.wantErr { if err == nil { t.Fatalf("expected error for %q", tc.uploadURL) @@ -63,7 +63,7 @@ func TestUploadCacheBackendSchemeDispatch(t *testing.T) { } func TestUploadCacheBackendEmptyURL(t *testing.T) { - backend, err := newUploadCacheBackend("", nil, t.TempDir(), slog.Default()) + backend, err := newUploadCacheBackend("", nil, t.TempDir(), nil, slog.Default()) if err != nil { t.Fatalf("unexpected error: %v", err) } diff --git a/spindle/engines/microvm/upload_cache_proxy.go b/spindle/engines/microvm/upload_cache_proxy.go index 9b9385d2..16153741 100644 --- a/spindle/engines/microvm/upload_cache_proxy.go +++ b/spindle/engines/microvm/upload_cache_proxy.go @@ -27,7 +27,7 @@ type UploadCacheProxy struct { backend UploadCacheBackend } -func StartUploadCacheProxy(ctx context.Context, cid uint32, uploadURL string, readUpstreams []CacheUpstream, stagingDir string, logger *slog.Logger) (*UploadCacheProxy, error) { +func StartUploadCacheProxy(ctx context.Context, cid uint32, uploadURL string, readUpstreams []CacheUpstream, stagingDir string, extraHeaders map[string]string, logger *slog.Logger) (*UploadCacheProxy, error) { if strings.TrimSpace(uploadURL) == "" { return nil, nil } @@ -37,7 +37,7 @@ func StartUploadCacheProxy(ctx context.Context, cid uint32, uploadURL string, re } logger = logger.With("where", "upload_cache_proxy", "cid", cid, "uploadURL", uploadURL) - backend, err := newUploadCacheBackend(uploadURL, readUpstreams, stagingDir, logger) + backend, err := newUploadCacheBackend(uploadURL, readUpstreams, stagingDir, extraHeaders, logger) if err != nil { return nil, err } @@ -73,7 +73,7 @@ func StartUploadCacheProxy(ctx context.Context, cid uint32, uploadURL string, re return proxy, nil } -func newUploadCacheBackend(uploadURL string, readUpstreams []CacheUpstream, stagingDir string, logger *slog.Logger) (UploadCacheBackend, error) { +func newUploadCacheBackend(uploadURL string, readUpstreams []CacheUpstream, stagingDir string, extraHeaders map[string]string, logger *slog.Logger) (UploadCacheBackend, error) { if strings.TrimSpace(uploadURL) == "" { return nil, nil } @@ -88,7 +88,7 @@ func newUploadCacheBackend(uploadURL string, readUpstreams []CacheUpstream, stag if target.Host == "" { return nil, fmt.Errorf("upload URL %q is missing host", uploadURL) } - return newHTTPUploadProxyBackend(target, readUpstreams, logger), nil + return newHTTPUploadProxyBackend(target, readUpstreams, extraHeaders, logger), nil case "ssh", "ssh-ng": return newNixStoreUploadBackend(target.String(), stagingDir, readUpstreams, logger, nil) diff --git a/spindle/engines/microvm/upload_cache_proxy_test.go b/spindle/engines/microvm/upload_cache_proxy_test.go index ef4e6405..31422ac0 100644 --- a/spindle/engines/microvm/upload_cache_proxy_test.go +++ b/spindle/engines/microvm/upload_cache_proxy_test.go @@ -35,7 +35,7 @@ func TestUploadProxyRewritesHostAndAuth(t *testing.T) { req := httptest.NewRequest(http.MethodPut, "http://127.0.0.1:10501/abc.narinfo", strings.NewReader("narinfo")) req.Host = "127.0.0.1:10501" rec := httptest.NewRecorder() - uploadProxyHandler(target, nil, slog.Default()).ServeHTTP(rec, req) + uploadProxyHandler(target, nil, nil, slog.Default()).ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) @@ -70,6 +70,7 @@ func TestUploadProxySkipsNarinfoAvailableUpstream(t *testing.T) { handler := uploadProxyHandler( mustParseURL(t, target.URL), []CacheUpstream{{url: mustParseURL(t, upstream.URL)}}, + nil, slog.Default(), ) @@ -98,6 +99,7 @@ func TestUploadProxyUploadsNarinfoNobodyHas(t *testing.T) { handler := uploadProxyHandler( mustParseURL(t, target.URL), []CacheUpstream{{url: mustParseURL(t, upstream.URL)}}, + nil, slog.Default(), ) @@ -116,7 +118,7 @@ func TestUploadProxySkipsNarinfoAlreadyOnTarget(t *testing.T) { })) defer target.Close() - handler := uploadProxyHandler(mustParseURL(t, target.URL), nil, slog.Default()) + handler := uploadProxyHandler(mustParseURL(t, target.URL), nil, nil, slog.Default()) req := httptest.NewRequest(http.MethodGet, "http://127.0.0.1:10501/abc.narinfo", nil) rec := httptest.NewRecorder() @@ -126,3 +128,29 @@ func TestUploadProxySkipsNarinfoAlreadyOnTarget(t *testing.T) { t.Fatalf("status: got %d, want 200", rec.Code) } } + +func TestUploadProxyInjectsExtraHeaders(t *testing.T) { + var gotAuth string + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + gotAuth = req.Header.Get("Authorization") + _, _ = io.WriteString(w, "ok") + })) + defer upstream.Close() + + target, err := url.Parse(upstream.URL) + if err != nil { + t.Fatal(err) + } + + extraHeaders := map[string]string{"Authorization": "Bearer token123"} + req := httptest.NewRequest(http.MethodPut, "http://127.0.0.1:10501/abc.narinfo", strings.NewReader("narinfo")) + rec := httptest.NewRecorder() + uploadProxyHandler(target, nil, extraHeaders, slog.Default()).ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String()) + } + if gotAuth != "Bearer token123" { + t.Fatalf("Authorization header: got %q, want %q", gotAuth, "Bearer token123") + } +} -- 2.51.2