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