1// SPDX-License-Identifier: MIT
2
3package resultssvc
4
5import (
6 "bytes"
7 "crypto/rand"
8 "crypto/sha256"
9 "encoding/base64"
10 "encoding/hex"
11 "encoding/json"
12 "errors"
13 "fmt"
14 "io"
15 "net/http"
16 "net/http/httptest"
17 "net/url"
18 "regexp"
19 "strconv"
20 "strings"
21 "sync/atomic"
22 "testing"
23 "time"
24
25 "connectrpc.com/connect"
26)
27
28func TestJobBackendIDAndScope(t *testing.T) {
29 sum := sha256.Sum256([]byte("job-1/2"))
30 if got, want := JobBackendID("job-1", 2), hex.EncodeToString(sum[:])[:32]; got != want {
31 t.Fatalf("JobBackendID = %q, want %q", got, want)
32 }
33 if !regexp.MustCompile(`^[0-9a-f]{32}$`).MatchString(JobBackendID("a:b", 1)) {
34 t.Fatal("not 32 lowercase hex characters")
35 }
36 if JobBackendID("job-1", 1) == JobBackendID("job-1", 2) {
37 t.Fatal("attempts must differ")
38 }
39 if got, want := Scope(42, "job-1", 2), "Actions.Results:42:"+JobBackendID("job-1", 2); got != want {
40 t.Fatalf("Scope = %q, want %q", got, want)
41 }
42 // The toolkit splits the scope on ':' and wants three parts.
43 if n := len(strings.Split(Scope(42, "a:b", 1), ":")); n != 3 {
44 t.Fatalf("scope has %d parts", n)
45 }
46}
47
48// Golden request bodies, as the real upload-artifact v4.6.2 / v7.0.2 and
49// download-artifact v4.3.0 sent them in the C0 spike.
50const (
51 goldenCreateV4 = `{"workflow_run_backend_id":"11111111-aaaa-4000-8000-000000000001","workflow_job_run_backend_id":"build","name":"single","version":4}`
52 goldenCreateV7 = `{"workflow_run_backend_id":"11111111-aaaa-4000-8000-000000000001","workflow_job_run_backend_id":"build","name":"report.html","version":7,"mime_type":"text/html","expires_at":"2026-11-14T12:00:00.000Z"}`
53 goldenFinalize = `{"workflow_run_backend_id":"11111111-aaaa-4000-8000-000000000001","workflow_job_run_backend_id":"build","name":"single","size":"146","hash":"%s"}`
54 goldenListName = `{"workflow_run_backend_id":"11111111-aaaa-4000-8000-000000000001","workflow_job_run_backend_id":"test","name_filter":"single"}`
55 goldenListID = `{"workflow_run_backend_id":"11111111-aaaa-4000-8000-000000000001","workflow_job_run_backend_id":"test","id_filter":"%s"}`
56 goldenListAll = `{"workflow_run_backend_id":"11111111-aaaa-4000-8000-000000000001","workflow_job_run_backend_id":"test"}`
57 goldenSignedURL = `{"workflow_run_backend_id":"11111111-aaaa-4000-8000-000000000001","workflow_job_run_backend_id":"build","name":"single"}`
58 goldenDelete = `{"workflow_run_backend_id":"11111111-aaaa-4000-8000-000000000001","workflow_job_run_backend_id":"build","name":"single"}`
59 goldenBlockList1 = `<?xml version="1.0" encoding="UTF-8" standalone="yes"?><BlockList><Latest>M2VhNWQ3MWYtZGEyYi00YmQ5LThiNDAtY2Y4MGM3OTBmY2NjMDAwMDAwMDAwMDAw</Latest></BlockList>`
60 goldenBlockID1 = "M2VhNWQ3MWYtZGEyYi00YmQ5LThiNDAtY2Y4MGM3OTBmY2NjMDAwMDAwMDAwMDAw"
61)
62
63func TestTwirpGolden(t *testing.T) {
64 r := newRig(t)
65 build := r.job(7, "build", 1)
66 test := r.job(7, "test", 1)
67 content := bytes.Repeat([]byte("zip!"), 37)[:146]
68
69 st, b := build.twirp(t, "CreateArtifact", goldenCreateV4)
70 var created map[string]any
71 if st != http.StatusOK || json.Unmarshal(b, &created) != nil {
72 t.Fatalf("create: %d %s", st, b)
73 }
74 if len(created) != 2 || created["ok"] != true {
75 t.Fatalf("create reply: %s", b)
76 }
77 u, err := url.Parse(created["signedUploadUrl"].(string))
78 if err != nil || u.Scheme != "http" || u.Host != strings.TrimPrefix(build.srv.URL, "http://") || u.Path != BlobPath+"upload/1" ||
79 u.Query().Get("se") != strconv.FormatInt(r.clk.now().Add(time.Hour).Unix(), 10) || u.Query().Get("sig") == "" {
80 t.Fatalf("signedUploadUrl: %v", created["signedUploadUrl"])
81 }
82 res, _ := do(t, http.MethodPut, u.String()+"&comp=block&blockid="+url.QueryEscape(goldenBlockID1), bytes.NewReader(content))
83 if res.StatusCode != http.StatusCreated {
84 t.Fatalf("block: %d", res.StatusCode)
85 }
86 res, _ = do(t, http.MethodPut, u.String()+"&comp=blocklist", strings.NewReader(goldenBlockList1))
87 if res.StatusCode != http.StatusCreated {
88 t.Fatalf("blocklist: %d", res.StatusCode)
89 }
90 st, b = build.twirp(t, "FinalizeArtifact", fmt.Sprintf(goldenFinalize, digestOf(content)))
91 if st != http.StatusOK || string(b) != `{"artifactId":"1","ok":true}`+"\n" {
92 t.Fatalf("finalize: %d %s", st, b)
93 }
94
95 wantItem := `{"workflowRunBackendId":"7","workflowJobRunBackendId":"` + JobBackendID("build", 1) +
96 `","databaseId":"1","name":"single","size":"146","createdAt":"2026-10-15T12:00:00Z","digest":"` + digestOf(content) + `"}`
97 for _, body := range []string{goldenListName, fmt.Sprintf(goldenListID, "1"), strings.Replace(fmt.Sprintf(goldenListID, "1"), `"1"`, "1", 1), goldenListAll} {
98 st, b = test.twirp(t, "ListArtifacts", body)
99 if st != http.StatusOK || string(b) != `{"artifacts":[`+wantItem+`]}`+"\n" {
100 t.Fatalf("list %s: %d %s", body, st, b)
101 }
102 }
103 st, b = test.twirp(t, "ListArtifacts", fmt.Sprintf(goldenListID, "2"))
104 if st != http.StatusOK || string(b) != `{"artifacts":[]}`+"\n" {
105 t.Fatalf("empty list: %d %s", st, b)
106 }
107
108 // The body names the uploader's job: ignored, the URL is for test's channel.
109 st, b = test.twirp(t, "GetSignedArtifactURL", goldenSignedURL)
110 var signed struct {
111 URL string `json:"signedUrl"`
112 }
113 if st != http.StatusOK || json.Unmarshal(b, &signed) != nil || !strings.HasPrefix(signed.URL, test.srv.URL+BlobPath+"download/1?se=") {
114 t.Fatalf("signed URL: %d %s", st, b)
115 }
116 res, got := do(t, http.MethodGet, signed.URL, nil)
117 if res.StatusCode != http.StatusOK || !bytes.Equal(got, content) {
118 t.Fatalf("download: %d %d bytes", res.StatusCode, len(got))
119 }
120
121 st, b = build.twirp(t, "DeleteArtifact", goldenDelete)
122 if st != http.StatusOK || string(b) != `{"artifactId":"1","ok":true}`+"\n" {
123 t.Fatalf("delete: %d %s", st, b)
124 }
125 st, b = test.twirp(t, "GetSignedArtifactURL", goldenSignedURL)
126 if st != http.StatusNotFound || !strings.Contains(string(b), `"code":"not_found"`) {
127 t.Fatalf("signed URL of a deleted artifact: %d %s", st, b)
128 }
129
130 // v7: mime_type and expires_at reach ermite.
131 st, b = build.twirp(t, "CreateArtifact", goldenCreateV7)
132 if st != http.StatusOK {
133 t.Fatalf("create v7: %d %s", st, b)
134 }
135 r.fake.mu.Lock()
136 a := r.fake.arts[2].a
137 r.fake.mu.Unlock()
138 if a.GetMediaType() != "text/html" || !a.GetExpiresAt().AsTime().Equal(time.Date(2026, 11, 14, 12, 0, 0, 0, time.UTC)) || a.GetJobId() != "build" || a.GetRunId() != 7 {
139 t.Fatalf("created: %v", a)
140 }
141 st, b = build.twirp(t, "CreateArtifact", `{"name":"x","expires_at":"tomorrow"}`)
142 if st != http.StatusBadRequest || !strings.Contains(string(b), `"invalid_argument"`) {
143 t.Fatalf("bad expires_at: %d %s", st, b)
144 }
145 st, b = build.twirp(t, "FinalizeArtifact", `{"name":"x","size":"many"}`)
146 if st != http.StatusBadRequest {
147 t.Fatalf("bad size: %d %s", st, b)
148 }
149 st, b = build.twirp(t, "ListArtifacts", `not json`)
150 if st != http.StatusBadRequest || !strings.Contains(string(b), `"malformed"`) {
151 t.Fatalf("bad JSON: %d %s", st, b)
152 }
153}
154
155func TestTwirpErrorMapping(t *testing.T) {
156 r := newRig(t)
157 j := r.job(1, "job", 1)
158 cases := []struct {
159 code connect.Code
160 msg string
161 status int
162 twirp string
163 want string
164 }{
165 {connect.CodeInvalidArgument, "bad name", 400, "invalid_argument", "bad name"},
166 {connect.CodeNotFound, "gone", 404, "not_found", "gone"},
167 {connect.CodeAlreadyExists, "", 409, "already_exists", msgExists},
168 {connect.CodeAlreadyExists, "taken", 409, "already_exists", "taken"},
169 {connect.CodeFailedPrecondition, "job ended", 412, "failed_precondition", "job ended"},
170 {connect.CodeResourceExhausted, "over the 10 GiB quota", 412, "failed_precondition", "over the 10 GiB quota"},
171 {connect.CodeUnauthenticated, "token", 401, "unauthenticated", "token"},
172 {connect.CodePermissionDenied, "denied", 401, "unauthenticated", "denied"},
173 {connect.CodeUnavailable, "down", 503, "unavailable", "down"},
174 {connect.CodeDeadlineExceeded, "slow", 503, "unavailable", "slow"},
175 {connect.CodeInternal, "bug", 503, "unavailable", "bug"},
176 {connect.CodeUnknown, "what", 503, "unavailable", "what"},
177 {connect.CodeCanceled, "stop", 503, "unavailable", "stop"},
178 {connect.CodeUnimplemented, "old", 503, "unavailable", "old"},
179 }
180 for _, c := range cases {
181 t.Run(c.code.String()+"/"+c.msg, func(t *testing.T) {
182 r.fake.mu.Lock()
183 r.fake.failNext["CreateJobArtifact"] = connect.NewError(c.code, errors.New(c.msg))
184 r.fake.mu.Unlock()
185 st, b := j.twirp(t, "CreateArtifact", goldenCreateV4)
186 var e map[string]string
187 if err := json.Unmarshal(b, &e); err != nil || st != c.status || e["code"] != c.twirp || e["msg"] != c.want || len(e) != 2 {
188 t.Fatalf("got %d %s, want %d %s %q", st, b, c.status, c.twirp, c.want)
189 }
190 })
191 }
192}
193
194func TestTwirpAuth(t *testing.T) {
195 r := newRig(t)
196 j := r.job(1, "job", 1)
197 other := r.job(1, "other", 1)
198 nextAttempt := Job{ID: "job", RunID: 1, Attempt: 2}
199 for name, tok := range map[string]string{
200 "no bearer": "",
201 "garbage": "abc",
202 "other job's token": other.token,
203 "no results scope": r.auth.mint("job", genericRead, time.Hour),
204 "scope of another run": r.auth.mint("job", Scope(2, "job", 1), time.Hour),
205 "scope of next try": r.auth.mint("job", Scope(nextAttempt.RunID, nextAttempt.ID, nextAttempt.Attempt), time.Hour),
206 "scope as a prefix": r.auth.mint("job", Scope(1, "job", 1)+"0", time.Hour),
207 "forged": (&jwtAuth{key: []byte("other"), now: r.clk.now}).mint("job", Scope(1, "job", 1), time.Hour),
208 "expired": r.auth.mint("job", Scope(1, "job", 1), -time.Second),
209 } {
210 st, b := j.twirpWith(t, tok, "ListArtifacts", goldenListAll)
211 if st != http.StatusUnauthorized || !strings.Contains(string(b), `"code":"unauthenticated"`) {
212 t.Errorf("%s: %d %s", name, st, b)
213 }
214 }
215 if st, b := j.twirp(t, "ListArtifacts", goldenListAll); st != http.StatusOK {
216 t.Fatalf("own token: %d %s", st, b)
217 }
218 if st, b := j.twirp(t, "Nope", "{}"); st != http.StatusNotFound || !strings.Contains(string(b), `"bad_route"`) {
219 t.Errorf("unknown method: %d %s", st, b)
220 }
221 res, b := do(t, http.MethodGet, j.srv.URL+TwirpPath+"ListArtifacts", nil)
222 if res.StatusCode != http.StatusMethodNotAllowed || !strings.Contains(string(b), `"bad_route"`) {
223 t.Errorf("GET: %d %s", res.StatusCode, b)
224 }
225 if res, _ := do(t, http.MethodGet, j.srv.URL+"/_apis/artifactcache/cache", nil); res.StatusCode != http.StatusNotFound {
226 t.Errorf("other path: %d", res.StatusCode)
227 }
228 st, b := j.twirp(t, "ListArtifacts", `{"name_filter":"`+strings.Repeat("x", maxTwirpBody)+`"}`)
229 if st != http.StatusBadRequest {
230 t.Errorf("huge body: %d %.100s", st, b)
231 }
232}
233
234func checkBlobError(t *testing.T, what string, res *http.Response, body []byte, status int, code string) {
235 t.Helper()
236 want := `<?xml version="1.0" encoding="utf-8"?><Error><Code>` + code + `</Code><Message>` + code + `</Message></Error>`
237 if res.StatusCode != status || res.Header.Get("x-ms-error-code") != code || string(body) != want {
238 t.Errorf("%s: %d %s %s, want %d %s", what, res.StatusCode, res.Header.Get("x-ms-error-code"), body, status, code)
239 }
240}
241
242func TestBlobUpload(t *testing.T) {
243 r := newRig(t)
244 j := r.job(1, "job", 1)
245 other := r.job(1, "other", 1)
246
247 t.Run("blocks", func(t *testing.T) {
248 u := j.create(t, "blocks", "")
249 res, b := do(t, http.MethodPut, u+"&comp=block&blockid="+url.QueryEscape(base64.StdEncoding.EncodeToString([]byte("b1"))), strings.NewReader("hello "))
250 if res.StatusCode != http.StatusCreated || len(b) != 0 || res.Header.Get("Content-Length") != "0" {
251 t.Fatalf("block: %d %q", res.StatusCode, b)
252 }
253 for _, h := range []string{"ETag", "Last-Modified", "x-ms-request-id", "x-ms-version", "x-ms-request-server-encrypted"} {
254 if res.Header.Get(h) == "" {
255 t.Errorf("no %s", h)
256 }
257 }
258 // A retry of a block replaces it.
259 do(t, http.MethodPut, u+"&comp=block&blockid="+base64.StdEncoding.EncodeToString([]byte("b2")), strings.NewReader("wrong"))
260 do(t, http.MethodPut, u+"&comp=block&blockid="+base64.StdEncoding.EncodeToString([]byte("b2")), strings.NewReader("world"))
261 list := `<BlockList><Uncommitted>` + base64.StdEncoding.EncodeToString([]byte("b1")) + `</Uncommitted><Latest>` +
262 base64.StdEncoding.EncodeToString([]byte("b2")) + `</Latest><Committed>` + base64.StdEncoding.EncodeToString([]byte("b1")) + `</Committed></BlockList>`
263 if res, b := do(t, http.MethodPut, u+"&comp=BlockList", strings.NewReader(list)); res.StatusCode != http.StatusCreated {
264 t.Fatalf("blocklist: %d %s", res.StatusCode, b)
265 }
266 id, _ := strconv.ParseInt(j.finalize(t, "blocks", []byte("hello worldhello ")), 10, 64)
267 if c, _ := r.fake.content(id); string(c) != "hello worldhello " {
268 t.Fatalf("content %q", c)
269 }
270 })
271
272 t.Run("empty block list", func(t *testing.T) {
273 u := j.create(t, "empty", "")
274 if res, b := do(t, http.MethodPut, u+"&comp=blocklist", strings.NewReader(`<?xml version="1.0" encoding="UTF-8" standalone="yes"?><BlockList/>`)); res.StatusCode != http.StatusCreated {
275 t.Fatalf("blocklist: %d %s", res.StatusCode, b)
276 }
277 id, _ := strconv.ParseInt(j.finalize(t, "empty", nil), 10, 64)
278 if c, ok := r.fake.content(id); !ok || len(c) != 0 {
279 t.Fatalf("content %q %v", c, ok)
280 }
281 })
282
283 t.Run("put blob", func(t *testing.T) {
284 u := j.create(t, "single-shot", "")
285 if res, b := do(t, http.MethodPut, u, strings.NewReader("all at once")); res.StatusCode != http.StatusCreated {
286 t.Fatalf("put blob: %d %s", res.StatusCode, b)
287 }
288 id, _ := strconv.ParseInt(j.finalize(t, "single-shot", []byte("all at once")), 10, 64)
289 if c, _ := r.fake.content(id); string(c) != "all at once" {
290 t.Fatalf("content %q", c)
291 }
292 u = j.create(t, "single-shot-empty", "")
293 if res, b := do(t, http.MethodPut, u, nil); res.StatusCode != http.StatusCreated {
294 t.Fatalf("empty put blob: %d %s", res.StatusCode, b)
295 }
296 j.finalize(t, "single-shot-empty", nil)
297 })
298
299 t.Run("refusals", func(t *testing.T) {
300 u := j.create(t, "refused", "")
301 pu, _ := url.Parse(u)
302 q := pu.Query()
303 blk := "&comp=block&blockid=" + base64.StdEncoding.EncodeToString([]byte("b"))
304 tamper := func(k, v string) string {
305 q2 := url.Values{"se": {q.Get("se")}, "sig": {q.Get("sig")}}
306 q2.Set(k, v)
307 return j.srv.URL + pu.Path + "?" + q2.Encode()
308 }
309 sig := []byte(q.Get("sig"))
310 sig[3] ^= 1
311 se, _ := strconv.ParseInt(q.Get("se"), 10, 64)
312 for name, target := range map[string]string{
313 "tampered signature": tamper("sig", string(sig)) + blk,
314 "no signature": j.srv.URL + pu.Path + "?se=" + q.Get("se") + blk,
315 "later expiry": tamper("se", strconv.FormatInt(se+1, 10)) + blk,
316 "another artifact": j.srv.URL + BlobPath + "upload/99?" + q.Encode() + blk,
317 "other job's URL": other.srv.URL + pu.Path + "?" + q.Encode() + blk,
318 "as a download URL": j.srv.URL + BlobPath + "download/" + strings.TrimPrefix(pu.Path, BlobPath+"upload/") + "?" + q.Encode(),
319 } {
320 method := http.MethodPut
321 if strings.Contains(target, "/download/") {
322 method = http.MethodGet
323 }
324 res, b := do(t, method, target, strings.NewReader("x"))
325 checkBlobError(t, name, res, b, http.StatusForbidden, "AuthenticationFailed")
326 }
327 res, b := do(t, http.MethodPut, u+"&comp=block&blockid=%%%", strings.NewReader("x"))
328 checkBlobError(t, "bad block id", res, b, http.StatusBadRequest, "InvalidQueryParameterValue")
329 res, b = do(t, http.MethodPut, u+"&comp=block&blockid="+base64.StdEncoding.EncodeToString(make([]byte, 65)), strings.NewReader("x"))
330 checkBlobError(t, "long block id", res, b, http.StatusBadRequest, "InvalidQueryParameterValue")
331 res, b = do(t, http.MethodPut, u+"&comp=block", strings.NewReader("x"))
332 checkBlobError(t, "no block id", res, b, http.StatusBadRequest, "InvalidQueryParameterValue")
333 res, b = do(t, http.MethodDelete, u, nil)
334 checkBlobError(t, "DELETE", res, b, http.StatusMethodNotAllowed, "UnsupportedHttpVerb")
335 res, b = do(t, http.MethodPut, u+"&comp=appendblock", strings.NewReader("x"))
336 checkBlobError(t, "append block", res, b, http.StatusMethodNotAllowed, "UnsupportedHttpVerb")
337 res, b = do(t, http.MethodPut, u+"&comp=blocklist", strings.NewReader(`<BlockList><Latest>`+base64.StdEncoding.EncodeToString([]byte("never"))+`</Latest></BlockList>`))
338 checkBlobError(t, "unknown block", res, b, http.StatusBadRequest, "InvalidBlockList")
339 // The large bodies go straight to the handler: over the loopback, a
340 // megabyte that the server reads in small parts can take seconds.
341 res, b = direct(t, r.svc.Handler(j.Job), u+"&comp=blocklist", strings.NewReader(`<BlockList><Latest>`+strings.Repeat(`AA==</Latest><Latest>`, maxBlocks)+`AA==</Latest></BlockList>`))
342 checkBlobError(t, "too many blocks", res, b, http.StatusBadRequest, "InvalidBlockList")
343 res, b = do(t, http.MethodPut, u+"&comp=blocklist", strings.NewReader(`<Other/>`))
344 checkBlobError(t, "not a block list", res, b, http.StatusBadRequest, "InvalidXmlDocument")
345 res, b = do(t, http.MethodPut, u+"&comp=blocklist", strings.NewReader(`<BlockList><Latest>AA==`))
346 checkBlobError(t, "truncated XML", res, b, http.StatusBadRequest, "InvalidXmlDocument")
347 res, b = direct(t, r.svc.Handler(j.Job), u+"&comp=blocklist", io.MultiReader(strings.NewReader(`<BlockList><!--`), bytes.NewReader(make([]byte, maxBlockListBody))))
348 checkBlobError(t, "huge block list", res, b, http.StatusRequestEntityTooLarge, "RequestBodyTooLarge")
349 res, b = do(t, http.MethodGet, j.srv.URL+BlobPath+"elsewhere/1", nil)
350 checkBlobError(t, "unknown op", res, b, http.StatusNotFound, "ResourceNotFound")
351
352 // The token's expiry ends the URL.
353 r.clk.add(time.Hour)
354 res, b = do(t, http.MethodPut, u+blk, strings.NewReader("x"))
355 checkBlobError(t, "expired", res, b, http.StatusForbidden, "AuthenticationFailed")
356 r.clk.add(-time.Hour)
357 })
358
359 t.Run("ermite errors", func(t *testing.T) {
360 u := j.create(t, "quota", "")
361 r.fake.mu.Lock()
362 r.fake.maxArtifact = 3 << 20
363 r.fake.mu.Unlock()
364 defer func() {
365 r.fake.mu.Lock()
366 r.fake.maxArtifact = 0
367 r.fake.mu.Unlock()
368 }()
369 res, b := do(t, http.MethodPut, u+"&comp=block&blockid=AA==", bytes.NewReader(make([]byte, 5<<20)))
370 checkBlobError(t, "over quota", res, b, http.StatusRequestEntityTooLarge, "RequestBodyTooLarge")
371 for _, c := range []struct {
372 code connect.Code
373 status int
374 azure string
375 }{
376 {connect.CodeNotFound, 404, "BlobNotFound"},
377 {connect.CodeFailedPrecondition, 400, "InvalidBlobOrBlock"},
378 {connect.CodeInvalidArgument, 400, "InvalidBlobOrBlock"},
379 {connect.CodeResourceExhausted, 413, "RequestBodyTooLarge"},
380 {connect.CodePermissionDenied, 403, "AuthenticationFailed"},
381 {connect.CodeUnauthenticated, 403, "AuthenticationFailed"},
382 {connect.CodeUnavailable, 503, "ServerBusy"},
383 {connect.CodeInternal, 503, "ServerBusy"},
384 } {
385 r.fake.mu.Lock()
386 r.fake.failNext["UploadJobArtifactBlock"] = connect.NewError(c.code, errors.New("injected"))
387 r.fake.mu.Unlock()
388 res, b := do(t, http.MethodPut, u+"&comp=block&blockid=AA==", strings.NewReader("x"))
389 checkBlobError(t, "block "+c.code.String(), res, b, c.status, c.azure)
390 }
391 r.fake.mu.Lock()
392 r.fake.failNext["CommitJobArtifact"] = connect.NewError(connect.CodeInvalidArgument, errors.New("injected"))
393 r.fake.mu.Unlock()
394 res, b = do(t, http.MethodPut, u+"&comp=blocklist", strings.NewReader("<BlockList/>"))
395 checkBlobError(t, "commit invalid_argument", res, b, http.StatusBadRequest, "InvalidBlockList")
396 // A different list after a commit: failed_precondition.
397 if res, _ := do(t, http.MethodPut, u+"&comp=blocklist", strings.NewReader("<BlockList/>")); res.StatusCode != http.StatusCreated {
398 t.Fatalf("commit: %d", res.StatusCode)
399 }
400 res, b = do(t, http.MethodPut, u+"&comp=blocklist", strings.NewReader("<BlockList><Latest>AA==</Latest></BlockList>"))
401 checkBlobError(t, "commit failed_precondition", res, b, http.StatusBadRequest, "InvalidBlockList")
402 res, b = do(t, http.MethodPut, u+"&comp=block&blockid=AA==", strings.NewReader("x"))
403 checkBlobError(t, "block after commit", res, b, http.StatusBadRequest, "InvalidBlobOrBlock")
404 })
405}
406
407// direct serves a PUT with h, without a network.
408func direct(t *testing.T, h http.Handler, target string, body io.Reader) (*http.Response, []byte) {
409 t.Helper()
410 rec := httptest.NewRecorder()
411 h.ServeHTTP(rec, httptest.NewRequest(http.MethodPut, target, body))
412 res := rec.Result()
413 b, _ := io.ReadAll(res.Body)
414 return res, b
415}
416
417// failingBody yields data, then fails.
418type failingBody struct {
419 data io.Reader
420}
421
422func (f *failingBody) Read(p []byte) (int, error) {
423 n, err := f.data.Read(p)
424 if err == io.EOF {
425 return n, errors.New("connection reset by sandbox")
426 }
427 return n, err
428}
429
430func TestBlockBodyErrorCancelsStream(t *testing.T) {
431 r := newRig(t)
432 j := r.job(1, "job", 1)
433 u, _ := url.Parse(j.create(t, "cut", ""))
434 id, _ := strconv.ParseInt(strings.TrimPrefix(u.Path, BlobPath+"upload/"), 10, 64)
435 h := r.svc.Handler(j.Job)
436 put := func(size int) *httptest.ResponseRecorder {
437 req := httptest.NewRequest(http.MethodPut, u.String()+"&comp=block&blockid=AA==", &failingBody{bytes.NewReader(make([]byte, size))})
438 rec := httptest.NewRecorder()
439 h.ServeHTTP(rec, req)
440 return rec
441 }
442 // The body fails after 2.5 MiB: two messages went to ermite, then the
443 // stream is cancelled.
444 if rec := put(5 << 19); rec.Code == http.StatusCreated {
445 t.Fatalf("a failed body got %d", rec.Code)
446 }
447 deadline := time.Now().Add(5 * time.Second)
448 for r.fake.streams() < 1 {
449 if time.Now().After(deadline) {
450 t.Fatal("ermite never saw the stream end")
451 }
452 time.Sleep(5 * time.Millisecond)
453 }
454 if n := r.fake.pendingBlocks(id); n != 0 {
455 t.Fatalf("ermite stored %d blocks of a broken upload", n)
456 }
457 // A body that fails before any byte never opens a stream.
458 if rec := put(0); rec.Code == http.StatusCreated {
459 t.Fatalf("a failed body got %d", rec.Code)
460 }
461 time.Sleep(50 * time.Millisecond)
462 if n := r.fake.streams(); n != 1 {
463 t.Fatalf("%d streams", n)
464 }
465 // And a good block is stored.
466 res, _ := do(t, http.MethodPut, u.String()+"&comp=block&blockid=AA==", bytes.NewReader(make([]byte, 5<<19)))
467 if res.StatusCode != http.StatusCreated || r.fake.pendingBlocks(id) != 1 {
468 t.Fatalf("good block: %d, %d blocks", res.StatusCode, r.fake.pendingBlocks(id))
469 }
470}
471
472func random(t *testing.T, n int) []byte {
473 t.Helper()
474 b := make([]byte, n)
475 if _, err := rand.Read(b); err != nil {
476 t.Fatal(err)
477 }
478 return b
479}
480
481func TestDownload(t *testing.T) {
482 r := newRig(t)
483 up := r.job(1, "up", 1)
484 down := r.job(1, "down", 1)
485 big := random(t, 3<<20+12345)
486 up.upload(t, "big", "", big, 8<<20)
487 up.upload(t, "report é.html", "text/html", []byte("<h1>raw</h1>"), 8<<20)
488 up.upload(t, "zero", "", nil, 8<<20)
489
490 t.Run("zip", func(t *testing.T) {
491 res, b := do(t, http.MethodGet, down.signedURL(t, "big"), nil)
492 if res.StatusCode != http.StatusOK || !bytes.Equal(b, big) || res.Header.Get("Content-Type") != "application/zip" ||
493 res.Header.Get("Content-Length") != strconv.Itoa(len(big)) || res.Header.Get("Content-Disposition") != "" {
494 t.Fatalf("%d %d bytes %v", res.StatusCode, len(b), res.Header)
495 }
496 })
497 t.Run("archive false", func(t *testing.T) {
498 res, b := do(t, http.MethodGet, down.signedURL(t, "report é.html"), nil)
499 if res.StatusCode != http.StatusOK || string(b) != "<h1>raw</h1>" || res.Header.Get("Content-Type") != "text/html" ||
500 res.Header.Get("Content-Disposition") != "attachment; filename*=UTF-8''report%20%C3%A9.html" {
501 t.Fatalf("%d %q %v", res.StatusCode, b, res.Header)
502 }
503 })
504 t.Run("empty", func(t *testing.T) {
505 res, b := do(t, http.MethodGet, down.signedURL(t, "zero"), nil)
506 if res.StatusCode != http.StatusOK || len(b) != 0 || res.Header.Get("Content-Length") != "0" {
507 t.Fatalf("%d %q %v", res.StatusCode, b, res.Header)
508 }
509 })
510 t.Run("not found", func(t *testing.T) {
511 u := down.signedURL(t, "big")
512 down.twirp(t, "DeleteArtifact", `{"name":"big"}`)
513 res, b := do(t, http.MethodGet, u, nil)
514 checkBlobError(t, "deleted", res, b, http.StatusNotFound, "BlobNotFound")
515 up.upload(t, "big", "", big, 8<<20)
516 })
517 t.Run("resume", func(t *testing.T) {
518 var cuts atomic.Int32
519 r.fake.mu.Lock()
520 r.fake.cut = func(open, sent int64) error {
521 // Each of the first two streams breaks after a chunk.
522 if sent >= 1<<20 && cuts.Load() < 2 {
523 cuts.Add(1)
524 return connect.NewError(connect.CodeUnavailable, errors.New("cut"))
525 }
526 return nil
527 }
528 r.fake.mu.Unlock()
529 res, b := do(t, http.MethodGet, down.signedURL(t, "big"), nil)
530 if res.StatusCode != http.StatusOK || !bytes.Equal(b, big) || cuts.Load() != 2 {
531 t.Fatalf("%d %d bytes of %d, %d cuts", res.StatusCode, len(b), len(big), cuts.Load())
532 }
533 })
534 t.Run("resume exhausted", func(t *testing.T) {
535 r.fake.mu.Lock()
536 r.fake.cut = func(open, sent int64) error {
537 if open > 0 || sent >= 1<<20 {
538 return connect.NewError(connect.CodeUnavailable, errors.New("cut"))
539 }
540 return nil
541 }
542 r.fake.mu.Unlock()
543 defer func() {
544 r.fake.mu.Lock()
545 r.fake.cut = nil
546 r.fake.mu.Unlock()
547 }()
548 res, err := http.Get(down.signedURL(t, "big"))
549 if err != nil {
550 t.Fatal(err)
551 }
552 defer func() { _ = res.Body.Close() }()
553 b, err := io.ReadAll(res.Body)
554 if res.StatusCode != http.StatusOK || err == nil || len(b) != 1<<20 {
555 t.Fatalf("%d, %d bytes, %v: want a truncated body", res.StatusCode, len(b), err)
556 }
557 })
558 t.Run("other run", func(t *testing.T) {
559 intruder := r.job(2, "intruder", 1)
560 st, b := intruder.twirp(t, "GetSignedArtifactURL", `{"name":"big"}`)
561 if st != http.StatusNotFound {
562 t.Fatalf("%d %s", st, b)
563 }
564 if st, b := intruder.twirp(t, "ListArtifacts", `{}`); st != http.StatusOK || string(b) != `{"artifacts":[]}`+"\n" {
565 t.Fatalf("%d %s", st, b)
566 }
567 // A URL of the run's artifact, signed for the intruder's channel (as if
568 // it had guessed the id), is still refused by ermite.
569 se := r.clk.now().Add(time.Hour).Unix()
570 u := intruder.srv.URL + BlobPath + "download/2?se=" + strconv.FormatInt(se, 10) + "&sig=" + r.svc.blobSig(intruder.Job, opDownload, 2, se)
571 res, b := do(t, http.MethodGet, u, nil)
572 checkBlobError(t, "other run", res, b, http.StatusNotFound, "BlobNotFound")
573 })
574}