1// SPDX-License-Identifier: MIT
2
3package cachesvc
4
5import (
6 "encoding/json"
7 "errors"
8 "fmt"
9 "io"
10 "net/http"
11 "net/url"
12 "os"
13 "strconv"
14 "strings"
15 "time"
16
17 "bnema.dev/ermite-runner/internal/registryproxy"
18)
19
20// APIPath is the prefix of the cache API: the toolkit appends "_apis/
21// artifactcache/<resource>" to ACTIONS_CACHE_URL
22// (toolkit/packages/cache/src/internal/cacheHttpClient.ts:47).
23const APIPath = "/_apis/artifactcache/"
24
25// Limits of the requests, from the toolkit (cache.ts:97-115: a key is at most
26// 512 characters and has no comma; restore keys are at most 10) and of this
27// service (the version is a sha256 hex string for the toolkit).
28const (
29 maxKeyLen = 512
30 maxVersionLen = 128
31 maxKeys = 11
32 maxJSONBody = 1 << 20
33)
34
35// Messages the toolkit looks for (cache.ts:50, constants.ts:45): a denial that
36// starts with them is logged as a warning, not as a failure.
37const (
38 writeDenied = "cache write denied: "
39 readDenied = "cache read denied: "
40)
41
42// Handler returns the HTTP handler of the cache service (API v1), for any job.
43func (s *Service) Handler() http.Handler { return s.handler("") }
44
45// handler is Handler for the channel of one job: with a job id, a token made
46// for another job is refused, so a token that left its sandbox is useless on
47// any other sandbox's channel.
48func (s *Service) handler(job string) http.Handler {
49 mux := http.NewServeMux()
50 if s.registry != nil {
51 registry := s.serveRegistry(job, make(chan struct{}, s.registryPerChannel), registryproxy.NewQuota(RegistryDownloadsPerChannel))
52 mux.HandleFunc("GET "+RegistryPath, registry)
53 mux.HandleFunc("HEAD "+RegistryPath, registry)
54 }
55 // The routes are those of the toolkit's v1 client (cacheHttpClient.ts):
56 // GET cache?keys=k1,k2&version=v :89 lookup
57 // GET caches?key=k :132 diagnostics (core.isDebug())
58 // POST caches {key,version,cacheSize} :190 reserve -> {cacheId}
59 // PATCH caches/{id} Content-Range :230 chunk upload
60 // POST caches/{id} {size} :330 commit
61 // GET <archiveLocation> downloadUtils.ts:168 (no bearer token)
62 mux.HandleFunc("GET "+APIPath+"cache", s.auth(job, s.lookup))
63 mux.HandleFunc("GET "+APIPath+"caches", s.auth(job, s.listCaches))
64 mux.HandleFunc("POST "+APIPath+"caches", s.auth(job, s.reserve))
65 mux.HandleFunc("PATCH "+APIPath+"caches/{id}", s.auth(job, s.upload))
66 mux.HandleFunc("POST "+APIPath+"caches/{id}", s.auth(job, s.commit))
67 mux.HandleFunc("GET "+APIPath+"artifacts/{id}", s.download(job))
68 mux.HandleFunc("HEAD "+APIPath+"artifacts/{id}", s.download(job))
69 mux.HandleFunc("GET "+ArchivePath+"{key}", s.serveActionArchive(job))
70 // net/http resets the write deadline per request only when the server has a
71 // WriteTimeout, which would cut long uploads and downloads: set a fresh
72 // progress deadline for each request here, so that a deadline left by the
73 // previous request of a kept-alive connection never applies to this one.
74 return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
75 rc := http.NewResponseController(w)
76 d := time.Now().Add(s.progress)
77 _ = rc.SetReadDeadline(d)
78 _ = rc.SetWriteDeadline(d)
79 mux.ServeHTTP(w, r)
80 })
81}
82
83// channelHandler is the handler of a sandbox's channel: the cache API, the
84// action archives and the registry, and when results is not nil everything else goes to it (the results service of the
85// job: Twirp and blob paths). The results service gets the same progress
86// deadlines as the cache: its bodies and replies must keep moving.
87func (s *Service) channelHandler(job string, results http.Handler) http.Handler {
88 cache := s.handler(job)
89 if results == nil {
90 return cache
91 }
92 return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
93 if strings.HasPrefix(r.URL.Path, APIPath) || strings.HasPrefix(r.URL.Path, ArchivePath) || strings.HasPrefix(r.URL.Path, RegistryPath) {
94 cache.ServeHTTP(w, r)
95 return
96 }
97 rc := http.NewResponseController(w)
98 pr := newProgressReader(r.Body, rc, s.progress, nil)
99 defer pr.done()
100 pr.set()
101 r.Body = readCloser{pr, r.Body}
102 results.ServeHTTP(&progressWriter{ResponseWriter: w, rc: rc, idle: s.progress}, r)
103 })
104}
105
106type handlerFunc func(w http.ResponseWriter, r *http.Request, c Claims)
107
108// auth checks the bearer token (ACTIONS_RUNTIME_TOKEN), and when job is not
109// empty that it was made for that job. Every refusal is a bare 403 whatever the
110// reason.
111func (s *Service) auth(job string, h handlerFunc) http.HandlerFunc {
112 return func(w http.ResponseWriter, r *http.Request) {
113 const p = "Bearer "
114 a := r.Header.Get("Authorization")
115 if len(a) <= len(p) || !strings.EqualFold(a[:len(p)], p) {
116 reply(w, http.StatusForbidden, "forbidden")
117 return
118 }
119 c, err := s.sign.verify(strings.TrimSpace(a[len(p):]), s.now())
120 if err != nil || (job != "" && c.JobID != job) {
121 reply(w, http.StatusForbidden, "forbidden")
122 return
123 }
124 // Every body, JSON or chunk, must keep moving: a client that stalls
125 // loses its connection (and an upload its in-flight range).
126 pr := newProgressReader(r.Body, http.NewResponseController(w), s.progress, s.remaining(c.Expiry))
127 defer pr.done()
128 r.Body = readCloser{pr, r.Body}
129 h(w, r, c)
130 }
131}
132
133type readCloser struct {
134 io.Reader
135 io.Closer
136}
137
138// reply sends the {"message": …} error body that @actions/http-client turns into
139// the error message.
140func reply(w http.ResponseWriter, code int, msg string) {
141 w.Header().Set("Content-Type", "application/json")
142 w.WriteHeader(code)
143 _ = json.NewEncoder(w).Encode(map[string]string{"message": msg})
144}
145
146func replyJSON(w http.ResponseWriter, code int, v any) {
147 w.Header().Set("Content-Type", "application/json")
148 w.WriteHeader(code)
149 _ = json.NewEncoder(w).Encode(v)
150}
151
152// ArtifactCacheEntry is the lookup answer (toolkit contracts.ts:10-16).
153type artifactCacheEntry struct {
154 CacheKey string `json:"cacheKey"`
155 Scope string `json:"scope"`
156 CacheVersion string `json:"cacheVersion"`
157 CreationTime string `json:"creationTime"`
158 ArchiveLocation string `json:"archiveLocation"`
159}
160
161func (s *Service) entryJSON(r *http.Request, c Claims, e Entry) artifactCacheEntry {
162 exp := c.Expiry
163 q := url.Values{"exp": {strconv.FormatInt(exp, 10)}, "job": {c.JobID}, "sig": {s.sign.archiveSig(e.ID, e.Repository, c.JobID, exp)}}
164 host := r.Host
165 return artifactCacheEntry{
166 CacheKey: e.Key, Scope: e.Ref, CacheVersion: e.Version,
167 CreationTime: e.Created.UTC().Format("2006-01-02T15:04:05Z"),
168 ArchiveLocation: "http://" + host + APIPath + "artifacts/" + strconv.FormatInt(e.ID, 10) + "?" + q.Encode(),
169 }
170}
171
172func validKey(k string) bool {
173 return k != "" && len(k) <= maxKeyLen && !strings.Contains(k, ",")
174}
175
176func validVersion(v string) bool { return v != "" && len(v) <= maxVersionLen }
177
178func (s *Service) lookup(w http.ResponseWriter, r *http.Request, c Claims) {
179 if len(c.ReadRefs) == 0 {
180 reply(w, http.StatusForbidden, readDenied+"this job may not read the cache")
181 return
182 }
183 q := r.URL.Query()
184 keys := strings.Split(q.Get("keys"), ",")
185 version := q.Get("version")
186 if len(keys) > maxKeys || !validVersion(version) {
187 reply(w, http.StatusBadRequest, "bad keys or version")
188 return
189 }
190 for _, k := range keys {
191 if !validKey(k) {
192 reply(w, http.StatusBadRequest, "bad keys or version")
193 return
194 }
195 }
196 e, ok := s.store.find(c.Repository, c.ReadRefs, keys, version)
197 if !ok {
198 w.WriteHeader(http.StatusNoContent)
199 return
200 }
201 replyJSON(w, http.StatusOK, s.entryJSON(r, c, e))
202}
203
204func (s *Service) listCaches(w http.ResponseWriter, r *http.Request, c Claims) {
205 if len(c.ReadRefs) == 0 {
206 reply(w, http.StatusForbidden, readDenied+"this job may not read the cache")
207 return
208 }
209 list := s.store.list(c.Repository, c.ReadRefs, r.URL.Query().Get("key"))
210 out := struct {
211 TotalCount int `json:"totalCount"`
212 ArtifactCaches []artifactCacheEntry `json:"artifactCaches"`
213 }{TotalCount: len(list), ArtifactCaches: []artifactCacheEntry{}}
214 for _, e := range list {
215 out.ArtifactCaches = append(out.ArtifactCaches, s.entryJSON(r, c, e))
216 }
217 replyJSON(w, http.StatusOK, out)
218}
219
220func (s *Service) reserve(w http.ResponseWriter, r *http.Request, c Claims) {
221 if c.WriteRef == "" {
222 reply(w, http.StatusForbidden, writeDenied+"this job may not write the cache")
223 return
224 }
225 var req struct {
226 Key string `json:"key"`
227 Version string `json:"version"`
228 CacheSize int64 `json:"cacheSize"`
229 }
230 if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxJSONBody)).Decode(&req); err != nil || !validKey(req.Key) || !validVersion(req.Version) || req.CacheSize < 0 {
231 reply(w, http.StatusBadRequest, "bad reservation")
232 return
233 }
234 id, err := s.store.reserve(c.JobID, c.Repository, c.WriteRef, req.Key, req.Version, req.CacheSize, time.Unix(c.Expiry, 0))
235 switch {
236 case errors.Is(err, errConflict):
237 reply(w, http.StatusConflict, err.Error())
238 case errors.Is(err, errTooBig):
239 reply(w, http.StatusBadRequest, err.Error())
240 case errors.Is(err, errNoRoom), errors.Is(err, errTooMany):
241 reply(w, http.StatusServiceUnavailable, err.Error())
242 case err != nil:
243 s.log.Error().Err(err).Msg("cache reserve")
244 reply(w, http.StatusInternalServerError, "cache service error")
245 default:
246 replyJSON(w, http.StatusCreated, map[string]int64{"cacheId": id})
247 }
248}
249
250func parseID(r *http.Request) (int64, bool) {
251 id, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
252 return id, err == nil && id > 0
253}
254
255// parseRange reads "bytes a-b/*" (cacheHttpClient.ts:206-212): both ends
256// inclusive; the total is a star or a positive size that covers the range.
257func parseRange(h string) (lo, hi int64, ok bool) {
258 rest, found := strings.CutPrefix(h, "bytes ")
259 if !found {
260 return 0, 0, false
261 }
262 span, total, found := strings.Cut(rest, "/")
263 if !found {
264 return 0, 0, false
265 }
266 a, b, found := strings.Cut(span, "-")
267 if !found {
268 return 0, 0, false
269 }
270 lo, err1 := strconv.ParseInt(a, 10, 64)
271 last, err2 := strconv.ParseInt(b, 10, 64)
272 if err1 != nil || err2 != nil || lo < 0 || last < lo || last >= 1<<62 {
273 return 0, 0, false
274 }
275 if total != "*" {
276 n, err := strconv.ParseInt(total, 10, 64)
277 if err != nil || n <= 0 || n < last+1 {
278 return 0, 0, false
279 }
280 }
281 return lo, last + 1, true
282}
283
284func (s *Service) upload(w http.ResponseWriter, r *http.Request, c Claims) {
285 id, ok := parseID(r)
286 lo, hi, ok2 := parseRange(r.Header.Get("Content-Range"))
287 if !ok || !ok2 {
288 reply(w, http.StatusBadRequest, "bad cache id or Content-Range")
289 return
290 }
291 path, err := s.store.beginChunk(id, c, lo, hi)
292 if err != nil {
293 s.storeError(w, err)
294 return
295 }
296 // The range counts as received only when the whole body was written and
297 // checked; on any failure its coverage stays removed.
298 written := false
299 defer func() { s.store.endChunk(id, lo, hi, written) }()
300 if s.hookChunk != nil {
301 s.hookChunk(id)
302 }
303 f, err := os.OpenFile(path, os.O_WRONLY, 0)
304 if err != nil {
305 s.log.Error().Err(err).Msg("cache chunk")
306 reply(w, http.StatusInternalServerError, "cache service error")
307 return
308 }
309 defer func() { _ = f.Close() }()
310 // Exactly the announced bytes: fewer or more is a broken client or a
311 // truncated connection.
312 n, err := io.Copy(io.NewOffsetWriter(f, lo), io.LimitReader(r.Body, hi-lo))
313 if err != nil || n != hi-lo {
314 reply(w, http.StatusBadRequest, fmt.Sprintf("the body is not the %d bytes of the Content-Range", hi-lo))
315 return
316 }
317 if extra, _ := io.CopyN(io.Discard, r.Body, 1); extra > 0 {
318 reply(w, http.StatusBadRequest, "the body is longer than the Content-Range")
319 return
320 }
321 if err := f.Close(); err != nil {
322 s.log.Error().Err(err).Msg("cache chunk")
323 reply(w, http.StatusInternalServerError, "cache service error")
324 return
325 }
326 written = true
327 w.WriteHeader(http.StatusNoContent)
328}
329
330// progressReader extends the read deadline of the connection before each read:
331// a client that sends nothing for the progress timeout loses its upload
332// (slowloris on bodies), a slow one that keeps sending does not.
333type progressReader struct {
334 r io.Reader
335 rc *http.ResponseController
336 idle time.Duration
337 remaining func() time.Duration
338}
339
340// remaining returns how long a token that expires at the Unix time exp (or a
341// signed archive URL) is still valid, by the service clock.
342func (s *Service) remaining(exp int64) func() time.Duration {
343 return func() time.Duration { return time.Unix(exp, 0).Sub(s.now()) }
344}
345
346// deadline is the next deadline of a transfer: the progress period from now, but
347// never later than the end of the credential that admitted it (a request
348// admitted before the expiry ends at the expiry). The clock of the credential is
349// the service's, so the remaining time is turned into a real-time deadline.
350func deadline(idle time.Duration, remaining func() time.Duration) time.Time {
351 d := idle
352 if remaining != nil {
353 d = max(0, min(d, remaining()))
354 }
355 return time.Now().Add(d)
356}
357
358func newProgressReader(r io.Reader, rc *http.ResponseController, idle time.Duration, remaining func() time.Duration) *progressReader {
359 return &progressReader{r: r, rc: rc, idle: idle, remaining: remaining}
360}
361
362// set moves the deadlines. The reply to a body that stalled still gets one more
363// progress period to be written, but not past the credential's end.
364func (p *progressReader) set() {
365 _ = p.rc.SetReadDeadline(deadline(p.idle, p.remaining))
366 _ = p.rc.SetWriteDeadline(deadline(2*p.idle, p.remaining))
367}
368
369// Read: reading is progress, so the reply can wait for the body as long as the
370// body keeps coming.
371func (p *progressReader) Read(b []byte) (int, error) {
372 p.set()
373 return p.r.Read(b)
374}
375
376// done leaves a deadline in place, never none: after the handler the server
377// still drains what is left of the body, and that must not wait for ever either.
378// The server resets the deadlines for the next request of the connection.
379func (p *progressReader) done() { p.set() }
380
381// progressWriter is the same for downloads: the write deadline moves with
382// every write.
383type progressWriter struct {
384 http.ResponseWriter
385 rc *http.ResponseController
386 idle time.Duration
387 remaining func() time.Duration
388}
389
390func (p *progressWriter) Write(b []byte) (int, error) {
391 _ = p.rc.SetWriteDeadline(deadline(p.idle, p.remaining))
392 return p.ResponseWriter.Write(b)
393}
394
395// Unwrap lets http.ResponseController reach the real writer.
396func (p *progressWriter) Unwrap() http.ResponseWriter { return p.ResponseWriter }
397
398var _ registryproxy.KeepAliver = (*progressWriter)(nil)
399
400// KeepAlive moves the deadlines without writing. The registry proxy calls it
401// while the client waits for a shared upstream download that is progressing,
402// before the first byte of the reply. The read deadline moves too: once it
403// passes, net/http cancels the request's context and the wait would end.
404func (p *progressWriter) KeepAlive() {
405 d := deadline(p.idle, p.remaining)
406 _ = p.rc.SetReadDeadline(d)
407 _ = p.rc.SetWriteDeadline(d)
408}
409
410// serveRegistry serves the Docker Hub mirror with the same sliding progress
411// deadline as the other downloads: a layer of GiB keeps going as long as it
412// moves, and a client that stops reading is cut after one progress period.
413// slots caps the requests of this channel in progress at once; one over the
414// cap gets 503 at once, which dockerd retries or answers by pulling from
415// Docker Hub, so the channel's connections stay free for the cache. quota is
416// the channel's share of the proxy's cold downloads. job, when not empty, is
417// the channel's job, which the registry logs.
418func (s *Service) serveRegistry(job string, slots chan struct{}, quota *registryproxy.Quota) http.HandlerFunc {
419 return func(w http.ResponseWriter, r *http.Request) {
420 select {
421 case slots <- struct{}{}:
422 defer func() { <-slots }()
423 default:
424 w.Header().Set("Retry-After", "1")
425 reply(w, http.StatusServiceUnavailable, "registry busy")
426 return
427 }
428 rc := http.NewResponseController(w)
429 defer func() { _ = rc.SetWriteDeadline(deadline(s.progress, nil)) }()
430 ctx := registryproxy.WithQuota(r.Context(), quota)
431 if job != "" {
432 ctx = registryproxy.WithJob(ctx, job)
433 }
434 r = r.WithContext(ctx)
435 s.registry.ServeHTTP(&progressWriter{ResponseWriter: w, rc: rc, idle: s.progress}, r)
436 }
437}
438
439func (s *Service) commit(w http.ResponseWriter, r *http.Request, c Claims) {
440 id, ok := parseID(r)
441 var req struct {
442 Size int64 `json:"size"`
443 }
444 if !ok || json.NewDecoder(http.MaxBytesReader(w, r.Body, maxJSONBody)).Decode(&req) != nil {
445 reply(w, http.StatusBadRequest, "bad cache id or size")
446 return
447 }
448 if err := s.store.commit(id, c, req.Size); err != nil {
449 s.storeError(w, err)
450 return
451 }
452 w.WriteHeader(http.StatusNoContent)
453}
454
455// storeError maps the errors of the store. A reservation of another job, ref or
456// repository is the same 403 as a bad token; an unknown or expired one is 404.
457func (s *Service) storeError(w http.ResponseWriter, err error) {
458 switch {
459 case errors.Is(err, errDenied):
460 reply(w, http.StatusForbidden, "forbidden")
461 case errors.Is(err, errNotFound):
462 reply(w, http.StatusNotFound, err.Error())
463 case errors.Is(err, errTooBig):
464 reply(w, http.StatusRequestEntityTooLarge, err.Error())
465 case errors.Is(err, errBadSize), errors.Is(err, errState):
466 reply(w, http.StatusBadRequest, err.Error())
467 case errors.Is(err, errOverlap):
468 reply(w, http.StatusConflict, err.Error())
469 case errors.Is(err, errNoRoom), errors.Is(err, errTooMany):
470 reply(w, http.StatusServiceUnavailable, err.Error())
471 default:
472 s.log.Error().Err(err).Msg("cache")
473 reply(w, http.StatusInternalServerError, "cache service error")
474 }
475}
476
477// download serves an archive at the location given by a lookup. The client
478// fetches it without the bearer token (downloadUtils.ts:168), so the URL
479// carries the proof: an HMAC over the entry, its repository, the job the URL
480// was issued to and an expiry. On the channel of a job (job not empty) a URL
481// issued to another job is refused, as a token of another job is.
482func (s *Service) download(job string) http.HandlerFunc {
483 return func(w http.ResponseWriter, r *http.Request) {
484 s.serveArchive(w, r, job)
485 }
486}
487
488func (s *Service) serveArchive(w http.ResponseWriter, r *http.Request, channelJob string) {
489 id, ok := parseID(r)
490 q := r.URL.Query()
491 exp, err := strconv.ParseInt(q.Get("exp"), 10, 64)
492 issued := q.Get("job")
493 if !ok || err != nil || issued == "" || (channelJob != "" && issued != channelJob) {
494 reply(w, http.StatusForbidden, "forbidden")
495 return
496 }
497 e, found := s.store.get(id)
498 if !found || !s.sign.archiveOK(id, e.Repository, issued, exp, q.Get("sig"), s.now()) {
499 reply(w, http.StatusForbidden, "forbidden")
500 return
501 }
502 f, e, release, err := s.store.open(id, e.Repository)
503 if err != nil {
504 reply(w, http.StatusNotFound, "no such cache")
505 return
506 }
507 // The entry cannot be evicted (its bytes stay charged) while it is read.
508 defer release()
509 defer func() { _ = f.Close() }()
510 rc := http.NewResponseController(w)
511 remaining := s.remaining(exp)
512 defer func() { _ = rc.SetWriteDeadline(deadline(s.progress, remaining)) }()
513 w.Header().Set("Content-Type", "application/octet-stream")
514 http.ServeContent(&progressWriter{ResponseWriter: w, rc: rc, idle: s.progress, remaining: remaining}, r, "", e.Created, f)
515}