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