alice.pds.demo.boltless.dev
This repository has no description
1package microvm
2
3import (
4 "context"
5 "crypto/rand"
6 "encoding/binary"
7 "errors"
8 "fmt"
9 "io"
10 "log/slog"
11 "net"
12 "net/http"
13 "net/http/httputil"
14 "net/url"
15 "strings"
16 "sync"
17 "time"
18
19 "github.com/mdlayher/vsock"
20
21 "tangled.org/core/spindle/netguard"
22)
23
24const (
25 readCacheProxyPortMin = 20000
26 readCacheProxyPortMax = 60000
27)
28
29type ReadCacheProxy struct {
30 port uint32
31
32 ln *vsock.Listener
33 server *http.Server
34}
35
36func StartReadCacheProxy(ctx context.Context, cid uint32, upstreams []CacheUpstream, logger *slog.Logger) (*ReadCacheProxy, error) {
37 if logger == nil {
38 logger = slog.Default()
39 }
40 logger = logger.With("where", "read_cache", "cid", cid)
41
42 if len(upstreams) == 0 {
43 return nil, nil
44 }
45
46 ln, port, err := listenRandomVsockPort(ctx)
47 if err != nil {
48 return nil, err
49 }
50
51 proxy := &ReadCacheProxy{
52 port: port,
53 ln: ln,
54 }
55 proxy.server = &http.Server{
56 Handler: cacheProxyHandler(upstreams, logger),
57 Protocols: cacheProxyProtocols(),
58 ReadHeaderTimeout: 10 * time.Second,
59 }
60
61 filtered := &cidFilteredVsockListener{
62 Listener: ln,
63 cid: cid,
64 logger: logger,
65 }
66 go func() {
67 if err := proxy.server.Serve(filtered); err != nil && !errors.Is(err, http.ErrServerClosed) && !errors.Is(err, net.ErrClosed) {
68 logger.Warn("proxy stopped", "cid", cid, "port", port, "error", err)
69 }
70 }()
71
72 logger.Info("started proxy", "cid", cid, "port", port, "upstreams", len(upstreams))
73 return proxy, nil
74}
75
76func (p *ReadCacheProxy) Port() uint32 {
77 if p == nil {
78 return 0
79 }
80 return p.port
81}
82
83func (p *ReadCacheProxy) Close() error {
84 if p == nil {
85 return nil
86 }
87
88 var closeErr error
89 if p.server != nil {
90 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
91 closeErr = errors.Join(closeErr, p.server.Shutdown(ctx))
92 cancel()
93 p.server = nil
94 }
95 if p.ln != nil {
96 closeErr = errors.Join(closeErr, p.ln.Close())
97 p.ln = nil
98 }
99 return closeErr
100}
101
102type cidFilteredVsockListener struct {
103 *vsock.Listener
104 cid uint32
105 logger *slog.Logger
106}
107
108func (l *cidFilteredVsockListener) Accept() (net.Conn, error) {
109 for {
110 conn, err := l.Listener.Accept()
111 if err != nil {
112 return nil, err
113 }
114
115 addr, ok := conn.RemoteAddr().(*vsock.Addr)
116 if ok && addr.ContextID == l.cid {
117 return conn, nil
118 }
119
120 l.logger.Warn("dropping proxy connection from unexpected cid", "remote", conn.RemoteAddr(), "expectedCID", l.cid)
121 _ = conn.Close()
122 }
123}
124
125func parseCacheUpstreams(raw []string) ([]*url.URL, error) {
126 upstreams := make([]*url.URL, 0, len(raw))
127 seen := make(map[string]struct{}, len(raw))
128 for _, value := range raw {
129 value = strings.TrimSpace(value)
130 if value == "" {
131 continue
132 }
133 if _, ok := seen[value]; ok {
134 continue
135 }
136 seen[value] = struct{}{}
137
138 parsed, err := url.Parse(value)
139 if err != nil {
140 return nil, fmt.Errorf("parse URL %q: %w", value, err)
141 }
142 if parsed.Scheme != "http" && parsed.Scheme != "https" {
143 return nil, fmt.Errorf("URL %q uses unsupported scheme %q", value, parsed.Scheme)
144 }
145 if parsed.Host == "" {
146 return nil, fmt.Errorf("URL %q is missing host", value)
147 }
148 upstreams = append(upstreams, parsed)
149 }
150 return upstreams, nil
151}
152
153type CacheUpstream struct {
154 url *url.URL
155 // guarded upstreams come from the workflow file
156 // requests to them are refused for special-purpose address ranges
157 guarded bool
158}
159
160func BuildCacheUpstreams(rawTrusted, rawGuarded []string) ([]CacheUpstream, error) {
161 trusted, err := parseCacheUpstreams(rawTrusted)
162 if err != nil {
163 return nil, err
164 }
165 guarded, err := parseCacheUpstreams(rawGuarded)
166 if err != nil {
167 return nil, err
168 }
169 return mergeCacheUpstreams(trusted, guarded), nil
170}
171
172func mergeCacheUpstreams(trusted, guarded []*url.URL) []CacheUpstream {
173 merged := make([]CacheUpstream, 0, len(trusted)+len(guarded))
174 seen := make(map[string]struct{}, len(trusted)+len(guarded))
175 for _, u := range trusted {
176 if _, ok := seen[u.String()]; ok {
177 continue
178 }
179 seen[u.String()] = struct{}{}
180 merged = append(merged, CacheUpstream{url: u})
181 }
182 for _, u := range guarded {
183 if _, ok := seen[u.String()]; ok {
184 continue
185 }
186 seen[u.String()] = struct{}{}
187 merged = append(merged, CacheUpstream{url: u, guarded: true})
188 }
189 return merged
190}
191
192func listenRandomVsockPort(ctx context.Context) (*vsock.Listener, uint32, error) {
193 var lastErr error
194 for range 32 {
195 port, err := randomVsockPort()
196 if err != nil {
197 return nil, 0, err
198 }
199 ln, err := vsock.ListenContextID(vsock.Host, port, nil)
200 if err == nil {
201 return ln, port, nil
202 }
203 lastErr = err
204
205 select {
206 case <-ctx.Done():
207 return nil, 0, ctx.Err()
208 default:
209 }
210 }
211 return nil, 0, fmt.Errorf("listen on random vsock port: %w", lastErr)
212}
213
214func randomVsockPort() (uint32, error) {
215 var data [4]byte
216 if _, err := rand.Read(data[:]); err != nil {
217 return 0, fmt.Errorf("allocate read vsock port: %w", err)
218 }
219 span := uint32(readCacheProxyPortMax - readCacheProxyPortMin)
220 return readCacheProxyPortMin + binary.BigEndian.Uint32(data[:])%span, nil
221}
222
223var proxyTransport = &http.Transport{
224 Proxy: http.ProxyFromEnvironment,
225 ForceAttemptHTTP2: true,
226 MaxIdleConns: 100,
227 IdleConnTimeout: 90 * time.Second,
228 TLSHandshakeTimeout: 10 * time.Second,
229 ExpectContinueTimeout: 1 * time.Second,
230}
231
232// for guarded upstreams, this will refuse requests made to blocked addresses
233var guardedProxyTransport = &http.Transport{
234 DialContext: (&net.Dialer{
235 Timeout: 30 * time.Second,
236 KeepAlive: 30 * time.Second,
237 Control: netguard.RefuseSpecialPurposeAddrs,
238 }).DialContext,
239 ForceAttemptHTTP2: true,
240 MaxIdleConns: 100,
241 IdleConnTimeout: 90 * time.Second,
242 TLSHandshakeTimeout: 10 * time.Second,
243 ExpectContinueTimeout: 1 * time.Second,
244}
245
246// the proxy is the cache as far as the guest is concerned, so we answer
247// /nix-cache-info ourselves instead of racing the upstreams for it. merging
248// those also doesn't make any sense (none of the options make sense for
249// merging)
250const nixCacheInfo = "StoreDir: /nix/store\nWantMassQuery: 1\nPriority: 40\n"
251
252func cacheProxyHandler(upstreams []CacheUpstream, logger *slog.Logger) http.Handler {
253 proxy := &httputil.ReverseProxy{
254 // nothing to do here: the racing transport builds the full URL per
255 // upstream, it just needs the guest's path/query left intact
256 Rewrite: func(*httputil.ProxyRequest) {},
257 ErrorLog: slog.NewLogLogger(logger.Handler(), slog.LevelError),
258 Transport: ¶llelRacingTransport{
259 upstreams: upstreams,
260 underlying: proxyTransport,
261 guardedUnderlying: guardedProxyTransport,
262 logger: logger,
263 },
264 }
265
266 return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
267 if r.URL.Path == "/nix-cache-info" {
268 w.Header().Set("Content-Type", "text/x-nix-cache-info")
269 _, _ = io.WriteString(w, nixCacheInfo)
270 return
271 }
272 proxy.ServeHTTP(w, r)
273 })
274}
275
276func cacheProxyProtocols() *http.Protocols {
277 protocols := new(http.Protocols)
278 protocols.SetHTTP1(true)
279 protocols.SetUnencryptedHTTP2(true)
280 return protocols
281}
282
283func mergeQuery(base, extra string) string {
284 switch {
285 case base == "":
286 return extra
287 case extra == "":
288 return base
289 default:
290 return base + "&" + extra
291 }
292}
293
294type parallelRacingTransport struct {
295 upstreams []CacheUpstream
296 underlying http.RoundTripper
297 guardedUnderlying http.RoundTripper
298 logger *slog.Logger
299}
300
301func (t *parallelRacingTransport) RoundTrip(req *http.Request) (*http.Response, error) {
302 type result struct {
303 resp *http.Response
304 err error
305 is404 bool
306 idx int
307 }
308
309 resCh := make(chan result, len(t.upstreams))
310 cancels := make([]context.CancelFunc, len(t.upstreams))
311 var wg sync.WaitGroup
312
313 for i, upstream := range t.upstreams {
314 wg.Add(1)
315 ctx, cancel := context.WithCancel(req.Context())
316 cancels[i] = cancel
317
318 go func(idx int, target CacheUpstream, uCtx context.Context) {
319 defer wg.Done()
320
321 raceReq := req.Clone(uCtx)
322 // rewrite to the target, joining the upstream's base path/query
323 // with what the guest asked for
324 raceReq.URL.Scheme = target.url.Scheme
325 raceReq.URL.Host = target.url.Host
326 raceReq.URL.Path = strings.TrimSuffix(target.url.Path, "/") + req.URL.Path
327 raceReq.URL.RawQuery = mergeQuery(target.url.RawQuery, req.URL.RawQuery)
328 // Host wins over URL.Host for the outgoing Host header, and the
329 // reverse proxy preserves the guest's (127.0.0.1:<port>), which
330 // host-routed upstreams like fastly reject with a 421
331 raceReq.Host = target.url.Host
332 // the transport doesn't turn URL userinfo into basic auth, only
333 // http.Client does, so do it ourselves
334 if user := target.url.User; user != nil {
335 password, _ := user.Password()
336 raceReq.SetBasicAuth(user.Username(), password)
337 }
338
339 rt := t.underlying
340 if target.guarded {
341 rt = t.guardedUnderlying
342 }
343 resp, err := rt.RoundTrip(raceReq)
344 if err != nil {
345 resCh <- result{err: err, idx: idx}
346 return
347 }
348 if resp.StatusCode == http.StatusNotFound {
349 _ = resp.Body.Close() // don't care about the body of a 404
350 resCh <- result{is404: true, idx: idx}
351 return
352 }
353 if resp.StatusCode >= 400 {
354 // an erroring upstream must not win over a healthy one
355 _ = resp.Body.Close()
356 resCh <- result{err: fmt.Errorf("upstream returned status %d", resp.StatusCode), idx: idx}
357 return
358 }
359 // yay, ok
360 resCh <- result{resp: resp, idx: idx}
361 }(i, upstream, ctx)
362 }
363
364 go func() {
365 wg.Wait()
366 close(resCh)
367 }()
368
369 var total404s int
370 for res := range resCh {
371 if res.is404 {
372 total404s++
373 if total404s == len(t.upstreams) {
374 for _, cancel := range cancels {
375 cancel()
376 }
377 return &http.Response{
378 StatusCode: http.StatusNotFound,
379 Body: io.NopCloser(strings.NewReader("404 nix path not found")),
380 Header: make(http.Header),
381 Request: req,
382 }, nil
383 }
384 continue
385 }
386
387 if res.err != nil {
388 if !errors.Is(res.err, context.Canceled) {
389 t.logger.Warn("upstream failed",
390 "path", req.URL.Path,
391 "error", res.err,
392 )
393 }
394 continue
395 }
396
397 // cancel other requests
398 for i, cancel := range cancels {
399 if i != res.idx {
400 cancel()
401 }
402 }
403 return res.resp, nil
404 }
405
406 for _, cancel := range cancels {
407 cancel()
408 }
409 return nil, errors.New("all upstreams failed or timed out")
410}