1// SPDX-License-Identifier: MIT
2
3package resultssvc
4
5import (
6 "context"
7 "crypto/hmac"
8 "encoding/base64"
9 "encoding/xml"
10 "errors"
11 "fmt"
12 "io"
13 "net/http"
14 "net/url"
15 "strconv"
16 "strings"
17 "time"
18
19 "bnema.dev/ermite-api/gen/ermite"
20 "connectrpc.com/connect"
21)
22
23// Blob operations: the first segment after BlobPath, and the first part of
24// the signature.
25const (
26 opUpload = "upload"
27 opDownload = "download"
28)
29
30// Limits of the blob requests.
31const (
32 // chunkSize is the largest data of one message to ermite
33 // (UploadJobArtifactBlockRequest.data, DownloadJobArtifactResponse.chunk),
34 // and the one buffer a block upload holds.
35 chunkSize = 1 << 20
36 // maxBlockIDLen is the largest block id, decoded (Azure's limit).
37 maxBlockIDLen = 64
38 // maxBlockListBody bounds a Put Block List body: 50,000 ids of 64 bytes in
39 // base64 with their tags is about 6 MiB.
40 maxBlockListBody = 8 << 20
41 // maxBlocks is the most ids a block list may name (Azure's limit, and
42 // CommitJobArtifactRequest's).
43 maxBlocks = 50000
44 // putBlobBlockID is the id of the one block of a single-shot Put Blob.
45 putBlobBlockID = "ermite-put-blob"
46 // downloadTries is how many times a download stream that breaks is reopened
47 // in a row without receiving a byte.
48 downloadTries = 3
49)
50
51// blobSig signs a blob URL: base64url(BlobMAC(op, id, job, se)). The SDK sends
52// no bearer token with blob requests (the URL is all it has), so the URL is the
53// credential. It is bound to the operation, the artifact, the job of the
54// channel and an expiry: a URL that leaks out of its sandbox is useless on the
55// channel of any other job.
56func (s *Service) blobSig(job Job, op string, id, se int64) string {
57 return base64.RawURLEncoding.EncodeToString(s.auth.BlobMAC([]byte(op), []byte(strconv.FormatInt(id, 10)), []byte(job.ID), []byte(strconv.FormatInt(se, 10))))
58}
59
60// blobOK checks the signature and the expiry of a blob URL, in constant time.
61func (s *Service) blobOK(job Job, op string, id int64, q url.Values) bool {
62 if s.auth == nil {
63 return false
64 }
65 se, err := strconv.ParseInt(q.Get("se"), 10, 64)
66 if err != nil || s.now().Unix() >= se {
67 return false
68 }
69 got, err := base64.RawURLEncoding.DecodeString(q.Get("sig"))
70 if err != nil {
71 return false
72 }
73 want, _ := base64.RawURLEncoding.DecodeString(s.blobSig(job, op, id, se))
74 return hmac.Equal(got, want)
75}
76
77// blob serves the Azure block blob subset that @azure/storage-blob uses on the
78// signed URLs: Put Block, Put Block List and Put Blob on upload URLs, Get Blob
79// on download URLs.
80func (s *Service) blob(job Job, w http.ResponseWriter, r *http.Request, rest string) {
81 op, idText, _ := strings.Cut(rest, "/")
82 id, err := strconv.ParseInt(idText, 10, 64)
83 if (op != opUpload && op != opDownload) || err != nil || id <= 0 {
84 writeBlobError(w, http.StatusNotFound, "ResourceNotFound")
85 return
86 }
87 q := r.URL.Query()
88 if !s.blobOK(job, op, id, q) {
89 writeBlobError(w, http.StatusForbidden, "AuthenticationFailed")
90 return
91 }
92 comp := q.Get("comp")
93 s.log.Debug().Str("method", r.Method).Str("op", op).Str("comp", comp).Int64("artifact", id).Str("job", job.ID).Msg("results blob")
94 switch {
95 case op == opUpload && r.Method == http.MethodPut && comp == "block":
96 bid, err := base64.StdEncoding.DecodeString(q.Get("blockid"))
97 if err != nil || len(bid) == 0 || len(bid) > maxBlockIDLen {
98 writeBlobError(w, http.StatusBadRequest, "InvalidQueryParameterValue")
99 return
100 }
101 if err := s.putBlock(r.Context(), job, id, bid, r.Body); err != nil {
102 s.blobFailed(w, err, "InvalidBlobOrBlock", id)
103 return
104 }
105 writeBlobCreated(w)
106 case op == opUpload && r.Method == http.MethodPut && strings.EqualFold(comp, "blocklist"):
107 ids, status, code := readBlockList(w, r)
108 if code != "" {
109 writeBlobError(w, status, code)
110 return
111 }
112 if err := s.commit(r.Context(), job, id, ids); err != nil {
113 s.blobFailed(w, err, "InvalidBlockList", id)
114 return
115 }
116 writeBlobCreated(w)
117 case op == opUpload && r.Method == http.MethodPut && comp == "":
118 // Put Blob: the whole content in one request (the SDK's uploadData for
119 // small buffers). It becomes one block, then a commit of that block.
120 bid := []byte(putBlobBlockID)
121 if err := s.putBlock(r.Context(), job, id, bid, r.Body); err != nil {
122 s.blobFailed(w, err, "InvalidBlobOrBlock", id)
123 return
124 }
125 if err := s.commit(r.Context(), job, id, [][]byte{bid}); err != nil {
126 s.blobFailed(w, err, "InvalidBlockList", id)
127 return
128 }
129 writeBlobCreated(w)
130 case op == opDownload && r.Method == http.MethodGet && comp == "":
131 s.download(job, w, r, id)
132 default:
133 writeBlobError(w, http.StatusMethodNotAllowed, "UnsupportedHttpVerb")
134 }
135}
136
137// errBody is a failure to read the request body: the client's side, not
138// ermite's.
139type errBody struct{ err error }
140
141func (e errBody) Error() string { return "reading the request body: " + e.err.Error() }
142func (e errBody) Unwrap() error { return e.err }
143
144// putBlock streams a request body to ermite as one block, in messages of at
145// most chunkSize. Nothing is spooled: memory is the one buffer. A body that
146// fails to read cancels the stream, so that ermite stores nothing of the block
147// (a stream that breaks stores nothing; one that ends cleanly would store a
148// short block).
149func (s *Service) putBlock(ctx context.Context, job Job, id int64, blockID []byte, body io.Reader) error {
150 ctx, cancel := context.WithCancel(ctx)
151 defer cancel()
152 buf := make([]byte, chunkSize)
153 var st *connect.ClientStreamForClient[ermite.UploadJobArtifactBlockRequest, ermite.UploadJobArtifactBlockResponse]
154 for {
155 n, rerr := io.ReadFull(body, buf)
156 if rerr != nil && !errors.Is(rerr, io.EOF) && !errors.Is(rerr, io.ErrUnexpectedEOF) {
157 if st != nil {
158 // Cancel first, then wait for the call to end without closing the
159 // request side: closing it would end the stream cleanly.
160 cancel()
161 if conn, err := st.Conn(); err == nil {
162 _ = conn.CloseResponse()
163 }
164 }
165 return errBody{rerr}
166 }
167 if n > 0 || st == nil {
168 msg := &ermite.UploadJobArtifactBlockRequest{Data: buf[:n]}
169 if st == nil {
170 // The stream opens on the first data read, so that a body that
171 // fails at once never reaches ermite.
172 st = job.Client.UploadJobArtifactBlock(ctx)
173 msg.JobId, msg.ArtifactId, msg.BlockId = job.ID, id, blockID
174 }
175 if err := st.Send(msg); err != nil {
176 // io.EOF: ermite ended the stream; its error comes with the
177 // response.
178 _, cerr := st.CloseAndReceive()
179 if cerr == nil {
180 cerr = err
181 }
182 return cerr
183 }
184 }
185 if rerr != nil {
186 break
187 }
188 }
189 _, err := st.CloseAndReceive()
190 return err
191}
192
193// commit sets the content of the artifact to the blocks named.
194func (s *Service) commit(ctx context.Context, job Job, id int64, ids [][]byte) error {
195 _, err := job.Client.CommitJobArtifact(ctx, connect.NewRequest(&ermite.CommitJobArtifactRequest{JobId: job.ID, ArtifactId: id, BlockIds: ids}))
196 return err
197}
198
199// readBlockList reads a Put Block List body: <BlockList> with <Latest>,
200// <Committed> or <Uncommitted> children holding base64 block ids. All three
201// name an uploaded block here: there is no committed blob to take a block
202// from. It returns a status and an Azure error code when the body is refused.
203// It reads tokens rather than decoding into a struct, so that a list over
204// maxBlocks is refused as soon as its id 50,001 opens.
205func readBlockList(w http.ResponseWriter, r *http.Request) ([][]byte, int, string) {
206 const (
207 badXML = "InvalidXmlDocument"
208 badBlock = "InvalidBlockList"
209 )
210 dec := xml.NewDecoder(http.MaxBytesReader(w, r.Body, maxBlockListBody))
211 var (
212 ids [][]byte
213 text []byte
214 depth int
215 root bool
216 )
217 for {
218 tok, err := dec.RawToken()
219 if errors.Is(err, io.EOF) {
220 break
221 }
222 if err != nil {
223 var mb *http.MaxBytesError
224 if errors.As(err, &mb) {
225 return nil, http.StatusRequestEntityTooLarge, "RequestBodyTooLarge"
226 }
227 return nil, http.StatusBadRequest, badXML
228 }
229 switch t := tok.(type) {
230 case xml.StartElement:
231 depth++
232 switch {
233 case depth == 1 && !root && t.Name.Local == "BlockList":
234 root = true
235 case depth == 2 && (t.Name.Local == "Latest" || t.Name.Local == "Committed" || t.Name.Local == "Uncommitted"):
236 if len(ids) == maxBlocks {
237 return nil, http.StatusBadRequest, badBlock
238 }
239 text = text[:0]
240 default:
241 return nil, http.StatusBadRequest, badXML
242 }
243 case xml.CharData:
244 if depth == 2 {
245 text = append(text, t...)
246 }
247 case xml.EndElement:
248 if depth == 2 {
249 id, err := base64.StdEncoding.DecodeString(strings.TrimSpace(string(text)))
250 if err != nil || len(id) == 0 || len(id) > maxBlockIDLen {
251 return nil, http.StatusBadRequest, badBlock
252 }
253 ids = append(ids, id)
254 }
255 depth--
256 }
257 }
258 if !root || depth != 0 {
259 return nil, http.StatusBadRequest, badXML
260 }
261 return ids, 0, ""
262}
263
264// download streams the content of an artifact. If ermite's stream breaks
265// after the headers, it is reopened at the offset reached and the same
266// response goes on; when that fails downloadTries times in a row the response
267// is aborted, so the client sees a truncated body and not a clean end.
268func (s *Service) download(job Job, w http.ResponseWriter, r *http.Request, id int64) {
269 ctx := r.Context()
270 open := func(off int64) (*connect.ServerStreamForClient[ermite.DownloadJobArtifactResponse], *ermite.DownloadJobArtifactResponse, error) {
271 st, err := job.Client.DownloadJobArtifact(ctx, connect.NewRequest(&ermite.DownloadJobArtifactRequest{JobId: job.ID, ArtifactId: id, Offset: off}))
272 if err != nil {
273 return nil, nil, err
274 }
275 if !st.Receive() {
276 err := st.Err()
277 _ = st.Close()
278 if err == nil {
279 err = connect.NewError(connect.CodeUnavailable, errors.New("the download stream ended before its first message"))
280 }
281 return nil, nil, err
282 }
283 first := st.Msg()
284 if first.GetArtifact() == nil || first.GetArtifact().GetId() != id {
285 _ = st.Close()
286 return nil, nil, connect.NewError(connect.CodeInternal, errors.New("the download stream does not describe the artifact"))
287 }
288 return st, first, nil
289 }
290 st, first, err := open(0)
291 if err != nil {
292 s.blobFailed(w, err, "InvalidQueryParameterValue", id)
293 return
294 }
295 a := first.GetArtifact()
296 size := a.GetSize()
297 h := w.Header()
298 mt := a.GetMediaType()
299 if mt == "" {
300 mt = "application/zip"
301 }
302 h.Set("Content-Type", mt)
303 if mt != "application/zip" {
304 // download-artifact v5+ names a file uploaded as is from it
305 // (streamExtractExternal, download-artifact v8.0.2 dist/index.js:126608-126618).
306 h.Set("Content-Disposition", "attachment; filename*=UTF-8''"+url.PathEscape(a.GetName()))
307 }
308 h.Set("Content-Length", strconv.FormatInt(size, 10))
309 h.Set("x-ms-blob-type", "BlockBlob")
310 if t := a.GetCreatedAt(); t != nil {
311 h.Set("Last-Modified", t.AsTime().UTC().Format(http.TimeFormat))
312 }
313 w.WriteHeader(http.StatusOK)
314
315 var off int64
316 msg := first
317 fails := 0
318 for {
319 for {
320 c := msg.GetChunk()
321 if off+int64(len(c)) > size {
322 s.log.Warn().Int64("artifact", id).Str("job", job.ID).Msg("results: the download stream is longer than the artifact")
323 _ = st.Close()
324 panic(http.ErrAbortHandler)
325 }
326 if len(c) > 0 {
327 if _, err := w.Write(c); err != nil {
328 _ = st.Close() // the client is gone
329 return
330 }
331 off += int64(len(c))
332 fails = 0
333 }
334 if !st.Receive() {
335 break
336 }
337 msg = st.Msg()
338 }
339 serr := st.Err()
340 _ = st.Close()
341 if off == size && serr == nil {
342 return
343 }
344 if serr == nil {
345 serr = fmt.Errorf("the stream ended at %d of %d bytes", off, size)
346 }
347 for {
348 fails++
349 if fails > downloadTries || ctx.Err() != nil {
350 s.log.Warn().Int64("artifact", id).Str("job", job.ID).Int64("offset", off).Err(serr).Msg("results: download aborted")
351 panic(http.ErrAbortHandler)
352 }
353 s.log.Debug().Int64("artifact", id).Int64("offset", off).Err(serr).Msg("results: download stream broke, reopening")
354 st, msg, serr = open(off)
355 if serr == nil {
356 break
357 }
358 }
359 }
360}
361
362// blobFailed answers an error of a blob request, from ermite or from the
363// body. invalid is the code of a 400 (Azure's code for a bad block or block
364// list).
365func (s *Service) blobFailed(w http.ResponseWriter, err error, invalid string, id int64) {
366 var be errBody
367 if errors.As(err, &be) {
368 var mb *http.MaxBytesError
369 if errors.As(err, &mb) {
370 writeBlobError(w, http.StatusRequestEntityTooLarge, "RequestBodyTooLarge")
371 return
372 }
373 s.log.Debug().Int64("artifact", id).Err(err).Msg("results: blob body")
374 writeBlobError(w, http.StatusBadRequest, "IncompleteBody")
375 return
376 }
377 code := connect.CodeOf(err)
378 s.log.Warn().Int64("artifact", id).Str("code", code.String()).Err(err).Msg("results: ermite refused")
379 switch code {
380 case connect.CodeResourceExhausted:
381 writeBlobError(w, http.StatusRequestEntityTooLarge, "RequestBodyTooLarge")
382 case connect.CodeNotFound:
383 writeBlobError(w, http.StatusNotFound, "BlobNotFound")
384 case connect.CodeFailedPrecondition, connect.CodeInvalidArgument:
385 writeBlobError(w, http.StatusBadRequest, invalid)
386 case connect.CodeUnauthenticated, connect.CodePermissionDenied:
387 writeBlobError(w, http.StatusForbidden, "AuthenticationFailed")
388 default:
389 writeBlobError(w, http.StatusServiceUnavailable, "ServerBusy")
390 }
391}
392
393// writeBlobCreated is Azure's 201 to a Put Block, Put Block List or Put Blob,
394// with the headers the SDK reads into its response object.
395func writeBlobCreated(w http.ResponseWriter) {
396 h := w.Header()
397 now := time.Now().UTC()
398 h.Set("ETag", fmt.Sprintf(`"0x%X"`, now.UnixNano()))
399 h.Set("Last-Modified", now.Format(http.TimeFormat))
400 h.Set("x-ms-request-id", strconv.FormatInt(now.UnixNano(), 16))
401 h.Set("x-ms-version", "2025-11-05")
402 h.Set("x-ms-request-server-encrypted", "false")
403 h.Set("Content-Length", "0")
404 w.WriteHeader(http.StatusCreated)
405}
406
407// writeBlobError is an Azure storage error: the code in the XML body and in
408// x-ms-error-code, which the SDK reads.
409func writeBlobError(w http.ResponseWriter, status int, code string) {
410 body := `<?xml version="1.0" encoding="utf-8"?><Error><Code>` + code + `</Code><Message>` + code + `</Message></Error>`
411 h := w.Header()
412 h.Set("Content-Type", "application/xml")
413 h.Set("x-ms-error-code", code)
414 h.Set("Content-Length", strconv.Itoa(len(body)))
415 w.WriteHeader(status)
416 _, _ = io.WriteString(w, body)
417}