Monorepo for Tangled
spindle/microvm: allow configuring headers for read/upload cache reqs
Signed-off-by: dawn <dawn@tangled.org>
+126
-33
+2
-2
cmd/spindle-microvm-run/main_linux.go
+2
-2
cmd/spindle-microvm-run/main_linux.go
···
190
190
}
191
191
defer conn.Close()
192
192
193
-
upstreams, err := microvm.BuildCacheUpstreams(cmd.StringSlice("cache-read-url"), nil)
193
+
upstreams, err := microvm.BuildCacheUpstreams(cmd.StringSlice("cache-read-url"), nil, nil)
194
194
if err != nil {
195
195
return fmt.Errorf("build cache upstreams: %w", err)
196
196
}
···
208
208
var uploadCache *microvm.UploadCacheProxy
209
209
if cmd.String("cache-upload-url") != "" {
210
210
var err error
211
-
uploadCache, err = microvm.StartUploadCacheProxy(ctx, vm.CID(), cmd.String("cache-upload-url"), upstreams, filepath.Join(vm.WorkDir(), "upload-cache"), logger)
211
+
uploadCache, err = microvm.StartUploadCacheProxy(ctx, vm.CID(), cmd.String("cache-upload-url"), upstreams, filepath.Join(vm.WorkDir(), "upload-cache"), nil, logger)
212
212
if err != nil {
213
213
return fmt.Errorf("start upload cache proxy: %w", err)
214
214
}
+14
docs/DOCS.md
+14
docs/DOCS.md
···
1501
1501
trusted public keys for those caches.
1502
1502
- `SPINDLE_NIX_CACHE_UPLOAD_URL`: Cache URL that paths built
1503
1503
in the guest are uploaded to.
1504
+
- `SPINDLE_NIX_CACHE_READ_REQUEST_HEADERS`: Extra HTTP headers
1505
+
sent with every read request to the operator-configured read
1506
+
caches. Format is comma-separated `Name:Value` pairs, split
1507
+
on the first colon. Headers set here are **not** forwarded
1508
+
to workflow-supplied caches (e.g. those declared under
1509
+
`caches:` in a workflow file). Useful for caches that might
1510
+
require auth (eg. [attic](https://github.com/zhaofengli/attic))
1511
+
Example:
1512
+
```
1513
+
SPINDLE_NIX_CACHE_READ_REQUEST_HEADERS="Authorization:Bearer mytoken"
1514
+
```
1515
+
- `SPINDLE_NIX_CACHE_UPLOAD_REQUEST_HEADERS`: Extra HTTP headers
1516
+
sent with every upload request to the upload cache. Same
1517
+
format as above.
1504
1518
1505
1519
### Running spindle
1506
1520
+6
-6
spindle/config/config.go
+6
-6
spindle/config/config.go
···
87
87
}
88
88
89
89
type NixCache struct {
90
-
ReadURLs []string `env:"READ_URLS"`
91
-
TrustedPublicKeys []string `env:"TRUSTED_PUBLIC_KEYS"`
92
-
UploadURL string `env:"UPLOAD_URL"`
90
+
ReadURLs []string `env:"READ_URLS"`
91
+
TrustedPublicKeys []string `env:"TRUSTED_PUBLIC_KEYS"`
92
+
UploadURL string `env:"UPLOAD_URL"`
93
+
UploadRequestHeaders map[string]string `env:"UPLOAD_REQUEST_HEADERS"`
94
+
ReadRequestHeaders map[string]string `env:"READ_REQUEST_HEADERS"`
93
95
}
94
96
95
97
type Config struct {
···
102
104
103
105
func Load(ctx context.Context) (*Config, error) {
104
106
var cfg Config
105
-
err := envconfig.Process(ctx, &cfg)
106
-
if err != nil {
107
+
if err := envconfig.Process(ctx, &cfg); err != nil {
107
108
return nil, err
108
109
}
109
-
110
110
return &cfg, nil
111
111
}
+3
-3
spindle/engines/microvm/engine.go
+3
-3
spindle/engines/microvm/engine.go
···
232
232
}
233
233
}()
234
234
235
-
upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs)
235
+
upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs, e.cfg.NixCache.ReadRequestHeaders)
236
236
if err != nil {
237
237
return err
238
238
}
···
242
242
}
243
243
state.ReadCache = readCache
244
244
stagingDir := filepath.Join(workDir, "upload-cache")
245
-
uploadCache, err := StartUploadCacheProxy(ctx, cid, e.cfg.NixCache.UploadURL, upstreams, stagingDir, l)
245
+
uploadCache, err := StartUploadCacheProxy(ctx, cid, e.cfg.NixCache.UploadURL, upstreams, stagingDir, e.cfg.NixCache.UploadRequestHeaders, l)
246
246
if err != nil {
247
247
return err
248
248
}
···
492
492
}
493
493
494
494
func (e *Engine) anyCacheHasPath(ctx context.Context, state *workflowState, storePath string) bool {
495
-
upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs)
495
+
upstreams, err := BuildCacheUpstreams(e.cfg.NixCache.ReadURLs, state.CacheReadURLs, e.cfg.NixCache.ReadRequestHeaders)
496
496
if err != nil {
497
497
e.l.Warn("config cache check: build upstreams failed; treating as absent", "path", storePath, "error", err)
498
498
return false
+9
-5
spindle/engines/microvm/read_cache_proxy.go
+9
-5
spindle/engines/microvm/read_cache_proxy.go
···
153
153
url *url.URL
154
154
// guarded upstreams come from the workflow file
155
155
// requests to them are refused for special-purpose address ranges
156
-
guarded bool
156
+
guarded bool
157
+
extraHeaders map[string]string
157
158
}
158
159
159
-
func BuildCacheUpstreams(rawTrusted, rawGuarded []string) ([]CacheUpstream, error) {
160
+
func BuildCacheUpstreams(rawTrusted, rawGuarded []string, extraHeaders map[string]string) ([]CacheUpstream, error) {
160
161
trusted, err := parseCacheUpstreams(rawTrusted)
161
162
if err != nil {
162
163
return nil, err
···
165
166
if err != nil {
166
167
return nil, err
167
168
}
168
-
return mergeCacheUpstreams(trusted, guarded), nil
169
+
return mergeCacheUpstreams(trusted, guarded, extraHeaders), nil
169
170
}
170
171
171
-
func mergeCacheUpstreams(trusted, guarded []*url.URL) []CacheUpstream {
172
+
func mergeCacheUpstreams(trusted, guarded []*url.URL, trustedHeaders map[string]string) []CacheUpstream {
172
173
merged := make([]CacheUpstream, 0, len(trusted)+len(guarded))
173
174
seen := make(map[string]struct{}, len(trusted)+len(guarded))
174
175
for _, u := range trusted {
···
176
177
continue
177
178
}
178
179
seen[u.String()] = struct{}{}
179
-
merged = append(merged, CacheUpstream{url: u})
180
+
merged = append(merged, CacheUpstream{url: u, extraHeaders: trustedHeaders})
180
181
}
181
182
for _, u := range guarded {
182
183
if _, ok := seen[u.String()]; ok {
···
351
352
if user := target.url.User; user != nil {
352
353
password, _ := user.Password()
353
354
raceReq.SetBasicAuth(user.Username(), password)
355
+
}
356
+
for name, value := range target.extraHeaders {
357
+
raceReq.Header.Set(name, value)
354
358
}
355
359
356
360
rt := t.underlying
+50
-6
spindle/engines/microvm/read_cache_proxy_test.go
+50
-6
spindle/engines/microvm/read_cache_proxy_test.go
···
28
28
29
29
req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil)
30
30
rec := httptest.NewRecorder()
31
-
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
31
+
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil, nil), slog.Default()).ServeHTTP(rec, req)
32
32
33
33
if rec.Code != http.StatusOK {
34
34
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
···
51
51
52
52
req := httptest.NewRequest(http.MethodGet, "http://guest/nix-cache-info", nil)
53
53
rec := httptest.NewRecorder()
54
-
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
54
+
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil, nil), slog.Default()).ServeHTTP(rec, req)
55
55
56
56
if rec.Code != http.StatusOK {
57
57
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
···
79
79
80
80
req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil)
81
81
rec := httptest.NewRecorder()
82
-
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
82
+
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil, nil), slog.Default()).ServeHTTP(rec, req)
83
83
84
84
if rec.Code != http.StatusOK {
85
85
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
···
112
112
113
113
req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil)
114
114
rec := httptest.NewRecorder()
115
-
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
115
+
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil, nil), slog.Default()).ServeHTTP(rec, req)
116
116
117
117
if rec.Code != http.StatusOK {
118
118
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
···
137
137
138
138
req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil)
139
139
rec := httptest.NewRecorder()
140
-
cacheProxyHandler(mergeCacheUpstreams(nil, upstreams), slog.Default()).ServeHTTP(rec, req)
140
+
cacheProxyHandler(mergeCacheUpstreams(nil, upstreams, nil), slog.Default()).ServeHTTP(rec, req)
141
141
142
142
if rec.Code != http.StatusBadGateway {
143
143
t.Fatalf("status: got %d, want 502; body=%q", rec.Code, rec.Body.String())
···
163
163
req := httptest.NewRequest(http.MethodGet, "http://127.0.0.1:10500/abc.narinfo", nil)
164
164
req.Host = "127.0.0.1:10500"
165
165
rec := httptest.NewRecorder()
166
-
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil), slog.Default()).ServeHTTP(rec, req)
166
+
cacheProxyHandler(mergeCacheUpstreams(upstreams, nil, nil), slog.Default()).ServeHTTP(rec, req)
167
167
168
168
if rec.Code != http.StatusOK {
169
169
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
170
170
}
171
171
}
172
+
173
+
func TestReadCacheProxyInjectsExtraHeadersForTrustedOnly(t *testing.T) {
174
+
var trustedGotAuth, guardedGotAuth string
175
+
176
+
trustedSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
177
+
trustedGotAuth = req.Header.Get("Authorization")
178
+
_, _ = io.WriteString(w, "from-trusted")
179
+
}))
180
+
defer trustedSrv.Close()
181
+
182
+
// verify guarded upstreams never receive the injected headers
183
+
guardedSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
184
+
guardedGotAuth = req.Header.Get("Authorization")
185
+
// return 404 so trusted wins the race
186
+
w.WriteHeader(http.StatusNotFound)
187
+
}))
188
+
defer guardedSrv.Close()
189
+
190
+
trustedURLs, err := parseCacheUpstreams([]string{trustedSrv.URL})
191
+
if err != nil {
192
+
t.Fatal(err)
193
+
}
194
+
guardedURLs, err := parseCacheUpstreams([]string{guardedSrv.URL})
195
+
if err != nil {
196
+
t.Fatal(err)
197
+
}
198
+
199
+
extraHeaders := map[string]string{"Authorization": "Bearer readtoken"}
200
+
handler := cacheProxyHandler(mergeCacheUpstreams(trustedURLs, guardedURLs, extraHeaders), slog.Default())
201
+
202
+
req := httptest.NewRequest(http.MethodGet, "http://guest/abc.narinfo", nil)
203
+
rec := httptest.NewRecorder()
204
+
handler.ServeHTTP(rec, req)
205
+
206
+
if rec.Code != http.StatusOK {
207
+
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
208
+
}
209
+
if trustedGotAuth != "Bearer readtoken" {
210
+
t.Fatalf("trusted Authorization: got %q, want %q", trustedGotAuth, "Bearer readtoken")
211
+
}
212
+
if guardedGotAuth != "" {
213
+
t.Fatalf("guarded Authorization should be empty, got %q", guardedGotAuth)
214
+
}
215
+
}
+6
-3
spindle/engines/microvm/upload_cache_http.go
+6
-3
spindle/engines/microvm/upload_cache_http.go
···
17
17
handler http.Handler
18
18
}
19
19
20
-
func newHTTPUploadProxyBackend(target *url.URL, readUpstreams []CacheUpstream, logger *slog.Logger) *httpUploadBackend {
21
-
return &httpUploadBackend{handler: uploadProxyHandler(target, readUpstreams, logger)}
20
+
func newHTTPUploadProxyBackend(target *url.URL, readUpstreams []CacheUpstream, extraHeaders map[string]string, logger *slog.Logger) *httpUploadBackend {
21
+
return &httpUploadBackend{handler: uploadProxyHandler(target, readUpstreams, extraHeaders, logger)}
22
22
}
23
23
24
24
func (b *httpUploadBackend) ServeHTTP(w http.ResponseWriter, r *http.Request) {
···
27
27
28
28
func (b *httpUploadBackend) Close() error { return nil }
29
29
30
-
func uploadProxyHandler(target *url.URL, readUpstreams []CacheUpstream, logger *slog.Logger) http.Handler {
30
+
func uploadProxyHandler(target *url.URL, readUpstreams []CacheUpstream, extraHeaders map[string]string, logger *slog.Logger) http.Handler {
31
31
rp := httputil.NewSingleHostReverseProxy(target)
32
32
rp.ErrorLog = slog.NewLogLogger(logger.Handler(), slog.LevelError)
33
33
···
41
41
if user := target.User; user != nil {
42
42
password, _ := user.Password()
43
43
req.SetBasicAuth(user.Username(), password)
44
+
}
45
+
for name, value := range extraHeaders {
46
+
req.Header.Set(name, value)
44
47
}
45
48
}
46
49
+2
-2
spindle/engines/microvm/upload_cache_nix_store_test.go
+2
-2
spindle/engines/microvm/upload_cache_nix_store_test.go
···
44
44
45
45
for _, tc := range cases {
46
46
t.Run(tc.uploadURL, func(t *testing.T) {
47
-
backend, err := newUploadCacheBackend(tc.uploadURL, nil, staging, logger)
47
+
backend, err := newUploadCacheBackend(tc.uploadURL, nil, staging, nil, logger)
48
48
if tc.wantErr {
49
49
if err == nil {
50
50
t.Fatalf("expected error for %q", tc.uploadURL)
···
63
63
}
64
64
65
65
func TestUploadCacheBackendEmptyURL(t *testing.T) {
66
-
backend, err := newUploadCacheBackend("", nil, t.TempDir(), slog.Default())
66
+
backend, err := newUploadCacheBackend("", nil, t.TempDir(), nil, slog.Default())
67
67
if err != nil {
68
68
t.Fatalf("unexpected error: %v", err)
69
69
}
+4
-4
spindle/engines/microvm/upload_cache_proxy.go
+4
-4
spindle/engines/microvm/upload_cache_proxy.go
···
27
27
backend UploadCacheBackend
28
28
}
29
29
30
-
func StartUploadCacheProxy(ctx context.Context, cid uint32, uploadURL string, readUpstreams []CacheUpstream, stagingDir string, logger *slog.Logger) (*UploadCacheProxy, error) {
30
+
func StartUploadCacheProxy(ctx context.Context, cid uint32, uploadURL string, readUpstreams []CacheUpstream, stagingDir string, extraHeaders map[string]string, logger *slog.Logger) (*UploadCacheProxy, error) {
31
31
if strings.TrimSpace(uploadURL) == "" {
32
32
return nil, nil
33
33
}
···
37
37
}
38
38
logger = logger.With("where", "upload_cache_proxy", "cid", cid, "uploadURL", uploadURL)
39
39
40
-
backend, err := newUploadCacheBackend(uploadURL, readUpstreams, stagingDir, logger)
40
+
backend, err := newUploadCacheBackend(uploadURL, readUpstreams, stagingDir, extraHeaders, logger)
41
41
if err != nil {
42
42
return nil, err
43
43
}
···
73
73
return proxy, nil
74
74
}
75
75
76
-
func newUploadCacheBackend(uploadURL string, readUpstreams []CacheUpstream, stagingDir string, logger *slog.Logger) (UploadCacheBackend, error) {
76
+
func newUploadCacheBackend(uploadURL string, readUpstreams []CacheUpstream, stagingDir string, extraHeaders map[string]string, logger *slog.Logger) (UploadCacheBackend, error) {
77
77
if strings.TrimSpace(uploadURL) == "" {
78
78
return nil, nil
79
79
}
···
88
88
if target.Host == "" {
89
89
return nil, fmt.Errorf("upload URL %q is missing host", uploadURL)
90
90
}
91
-
return newHTTPUploadProxyBackend(target, readUpstreams, logger), nil
91
+
return newHTTPUploadProxyBackend(target, readUpstreams, extraHeaders, logger), nil
92
92
93
93
case "ssh", "ssh-ng":
94
94
return newNixStoreUploadBackend(target.String(), stagingDir, readUpstreams, logger, nil)
+30
-2
spindle/engines/microvm/upload_cache_proxy_test.go
+30
-2
spindle/engines/microvm/upload_cache_proxy_test.go
···
35
35
req := httptest.NewRequest(http.MethodPut, "http://127.0.0.1:10501/abc.narinfo", strings.NewReader("narinfo"))
36
36
req.Host = "127.0.0.1:10501"
37
37
rec := httptest.NewRecorder()
38
-
uploadProxyHandler(target, nil, slog.Default()).ServeHTTP(rec, req)
38
+
uploadProxyHandler(target, nil, nil, slog.Default()).ServeHTTP(rec, req)
39
39
40
40
if rec.Code != http.StatusOK {
41
41
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
···
70
70
handler := uploadProxyHandler(
71
71
mustParseURL(t, target.URL),
72
72
[]CacheUpstream{{url: mustParseURL(t, upstream.URL)}},
73
+
nil,
73
74
slog.Default(),
74
75
)
75
76
···
98
99
handler := uploadProxyHandler(
99
100
mustParseURL(t, target.URL),
100
101
[]CacheUpstream{{url: mustParseURL(t, upstream.URL)}},
102
+
nil,
101
103
slog.Default(),
102
104
)
103
105
···
116
118
}))
117
119
defer target.Close()
118
120
119
-
handler := uploadProxyHandler(mustParseURL(t, target.URL), nil, slog.Default())
121
+
handler := uploadProxyHandler(mustParseURL(t, target.URL), nil, nil, slog.Default())
120
122
121
123
req := httptest.NewRequest(http.MethodGet, "http://127.0.0.1:10501/abc.narinfo", nil)
122
124
rec := httptest.NewRecorder()
···
126
128
t.Fatalf("status: got %d, want 200", rec.Code)
127
129
}
128
130
}
131
+
132
+
func TestUploadProxyInjectsExtraHeaders(t *testing.T) {
133
+
var gotAuth string
134
+
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
135
+
gotAuth = req.Header.Get("Authorization")
136
+
_, _ = io.WriteString(w, "ok")
137
+
}))
138
+
defer upstream.Close()
139
+
140
+
target, err := url.Parse(upstream.URL)
141
+
if err != nil {
142
+
t.Fatal(err)
143
+
}
144
+
145
+
extraHeaders := map[string]string{"Authorization": "Bearer token123"}
146
+
req := httptest.NewRequest(http.MethodPut, "http://127.0.0.1:10501/abc.narinfo", strings.NewReader("narinfo"))
147
+
rec := httptest.NewRecorder()
148
+
uploadProxyHandler(target, nil, extraHeaders, slog.Default()).ServeHTTP(rec, req)
149
+
150
+
if rec.Code != http.StatusOK {
151
+
t.Fatalf("status: got %d, want 200; body=%q", rec.Code, rec.Body.String())
152
+
}
153
+
if gotAuth != "Bearer token123" {
154
+
t.Fatalf("Authorization header: got %q, want %q", gotAuth, "Bearer token123")
155
+
}
156
+
}