1// SPDX-License-Identifier: MIT
2
3package resultssvc
4
5import (
6 "bytes"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "io"
11 "net/http"
12 "net/url"
13 "strconv"
14 "strings"
15 "time"
16
17 "bnema.dev/ermite-api/gen/ermite"
18 "connectrpc.com/connect"
19 "google.golang.org/protobuf/types/known/timestamppb"
20)
21
22// maxTwirpBody bounds a Twirp request body: the largest the toolkit sends is a
23// few hundred bytes.
24const maxTwirpBody = 1 << 20
25
26// msgExists is the message of an already_exists reply when ermite gives none:
27// the toolkit's own wording for it.
28const msgExists = "an artifact with this name already exists on the workflow run"
29
30// twirpError is an error reply: a status and the {"code","msg"} body of Twirp.
31// The toolkit retries 5xx and 429 (ArtifactHttpClient.isRetryableHttpStatusCode,
32// upload-artifact v4.6.2 dist/upload/index.js:2827-2837) and fails at once on
33// anything else, with msg in the job log (:2787-2793). The body must be JSON
34// whatever the status: a body that does not parse is retried as a network
35// error (:2795-2806).
36type twirpError struct {
37 status int
38 code string
39 msg string
40}
41
42func (e *twirpError) Error() string { return e.code + ": " + e.msg }
43
44// twirpRequest is the union of the request messages of the methods. The
45// toolkit's generated client sends proto field names (useProtoFieldName),
46// wrapper types (StringValue, Int64Value) as bare values and int64 as strings;
47// the camelCase names that protojson also accepts are read too. Unknown fields
48// are ignored. The run and job backend ids are not read: everything comes from
49// the Job of the channel.
50type twirpRequest struct {
51 Name string `json:"name"`
52 MimeType string `json:"mime_type"`
53 MimeTypeCC string `json:"mimeType"`
54 ExpiresAt string `json:"expires_at"`
55 ExpiresAtCC string `json:"expiresAt"`
56 Size json.RawMessage `json:"size"`
57 Hash string `json:"hash"`
58 NameFilter string `json:"name_filter"`
59 NameFilterCC string `json:"nameFilter"`
60 IDFilter json.RawMessage `json:"id_filter"`
61 IDFilterCC json.RawMessage `json:"idFilter"`
62}
63
64func firstOf[T string | json.RawMessage](a, b T) T {
65 if len(a) > 0 {
66 return a
67 }
68 return b
69}
70
71// int64Value reads an int64 sent as a JSON string (protojson) or a number. An
72// absent or null value is 0.
73func int64Value(raw json.RawMessage, field string) (int64, *twirpError) {
74 v := strings.TrimSpace(string(raw))
75 if v == "" || v == "null" {
76 return 0, nil
77 }
78 if s, err := strconv.Unquote(v); err == nil {
79 v = s
80 }
81 n, err := strconv.ParseInt(v, 10, 64)
82 if err != nil {
83 return 0, &twirpError{http.StatusBadRequest, "invalid_argument", field + " is not an int64"}
84 }
85 return n, nil
86}
87
88// artifactJSON is an item of ListArtifactsResponse, in the camelCase of the
89// toolkit's fromJson with int64 as strings.
90type artifactJSON struct {
91 WorkflowRunBackendID string `json:"workflowRunBackendId"`
92 WorkflowJobRunBackendID string `json:"workflowJobRunBackendId"`
93 DatabaseID string `json:"databaseId"`
94 Name string `json:"name"`
95 Size string `json:"size"`
96 CreatedAt string `json:"createdAt,omitempty"`
97 Digest string `json:"digest,omitempty"`
98}
99
100// twirp serves one Twirp method. The routes are those of the toolkit's
101// ArtifactServiceClientJSON: CreateArtifact, FinalizeArtifact, ListArtifacts,
102// GetSignedArtifactURL and DeleteArtifact, all POST with a JSON body.
103func (s *Service) twirp(job Job, w http.ResponseWriter, r *http.Request, method string) {
104 if r.Method != http.MethodPost {
105 writeTwirpError(w, &twirpError{http.StatusMethodNotAllowed, "bad_route", "unsupported method " + r.Method})
106 return
107 }
108 var call func(Job, *http.Request, *twirpRequest, time.Time) (any, error)
109 switch method {
110 case "CreateArtifact":
111 call = s.createArtifact
112 case "FinalizeArtifact":
113 call = s.finalizeArtifact
114 case "ListArtifacts":
115 call = s.listArtifacts
116 case "GetSignedArtifactURL":
117 call = s.signedArtifactURL
118 case "DeleteArtifact":
119 call = s.deleteArtifact
120 default:
121 writeTwirpError(w, &twirpError{http.StatusNotFound, "bad_route", "no such method: " + method})
122 return
123 }
124 expiry, ok := s.authorize(job, r)
125 if !ok {
126 writeTwirpError(w, &twirpError{http.StatusUnauthorized, "unauthenticated", "invalid runtime token"})
127 return
128 }
129 body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxTwirpBody))
130 if err != nil {
131 writeTwirpError(w, &twirpError{http.StatusBadRequest, "malformed", "the request body could not be read"})
132 return
133 }
134 var req twirpRequest
135 if err := json.Unmarshal(body, &req); err != nil {
136 writeTwirpError(w, &twirpError{http.StatusBadRequest, "malformed", "the request body is not a JSON " + method + " request"})
137 return
138 }
139 s.log.Debug().Str("method", method).Str("job", job.ID).Str("name", req.Name).Msg("results twirp")
140 out, err := call(job, r, &req, expiry)
141 if err != nil {
142 var te *twirpError
143 if !errors.As(err, &te) {
144 te = twirpFromConnect(err)
145 s.log.Warn().Str("method", method).Str("job", job.ID).Str("code", connect.CodeOf(err).String()).Str("error", te.msg).Msg("results: ermite refused")
146 }
147 writeTwirpError(w, te)
148 return
149 }
150 w.Header().Set("Content-Type", "application/json")
151 w.WriteHeader(http.StatusOK)
152 _ = json.NewEncoder(w).Encode(out)
153}
154
155// authorize checks the bearer token: a runtime token of this channel's job
156// whose scopes hold the job's Actions.Results scope. It returns the token's
157// expiry, which is the expiry of the blob URLs. Every refusal looks the same.
158func (s *Service) authorize(job Job, r *http.Request) (time.Time, bool) {
159 const p = "Bearer "
160 a := r.Header.Get("Authorization")
161 if len(a) <= len(p) || !strings.EqualFold(a[:len(p)], p) || s.auth == nil {
162 return time.Time{}, false
163 }
164 jobID, scopes, expiry, err := s.auth.RuntimeToken(strings.TrimSpace(a[len(p):]))
165 if err != nil || jobID != job.ID || !s.now().Before(expiry) {
166 return time.Time{}, false
167 }
168 want := Scope(job.RunID, job.ID, job.Attempt)
169 for _, sc := range strings.Fields(scopes) {
170 if sc == want {
171 return expiry, true
172 }
173 }
174 return time.Time{}, false
175}
176
177func (s *Service) createArtifact(job Job, r *http.Request, req *twirpRequest, expiry time.Time) (any, error) {
178 in := &ermite.CreateJobArtifactRequest{JobId: job.ID, Name: req.Name, MediaType: firstOf(req.MimeType, req.MimeTypeCC)}
179 if v := firstOf(req.ExpiresAt, req.ExpiresAtCC); v != "" {
180 t, err := time.Parse(time.RFC3339Nano, v)
181 if err != nil {
182 return nil, &twirpError{http.StatusBadRequest, "invalid_argument", "expires_at is not an RFC 3339 time"}
183 }
184 in.ExpiresAt = timestamppb.New(t)
185 }
186 res, err := job.Client.CreateJobArtifact(r.Context(), connect.NewRequest(in))
187 if err != nil {
188 return nil, err
189 }
190 id := res.Msg.GetArtifactId()
191 return map[string]any{"ok": true, "signedUploadUrl": s.blobURL(job, r, opUpload, id, expiry)}, nil
192}
193
194func (s *Service) finalizeArtifact(job Job, r *http.Request, req *twirpRequest, _ time.Time) (any, error) {
195 size, te := int64Value(req.Size, "size")
196 if te != nil {
197 return nil, te
198 }
199 res, err := job.Client.FinalizeJobArtifact(r.Context(), connect.NewRequest(&ermite.FinalizeJobArtifactRequest{
200 JobId: job.ID, Name: req.Name, Size: size, Digest: req.Hash,
201 }))
202 if err != nil {
203 return nil, err
204 }
205 return map[string]any{"ok": true, "artifactId": strconv.FormatInt(res.Msg.GetArtifact().GetId(), 10)}, nil
206}
207
208func (s *Service) listArtifacts(job Job, r *http.Request, req *twirpRequest, _ time.Time) (any, error) {
209 id, te := int64Value(firstOf(req.IDFilter, req.IDFilterCC), "id_filter")
210 if te != nil {
211 return nil, te
212 }
213 res, err := job.Client.ListJobArtifacts(r.Context(), connect.NewRequest(&ermite.ListJobArtifactsRequest{
214 JobId: job.ID, Name: firstOf(req.NameFilter, req.NameFilterCC), ArtifactId: id,
215 }))
216 if err != nil {
217 return nil, err
218 }
219 items := make([]artifactJSON, 0, len(res.Msg.GetArtifacts()))
220 for _, a := range res.Msg.GetArtifacts() {
221 it := artifactJSON{
222 WorkflowRunBackendID: strconv.FormatInt(job.RunID, 10),
223 // The job of the artifact, not the caller: download-artifact passes
224 // it back in GetSignedArtifactURL, which ignores it.
225 WorkflowJobRunBackendID: JobBackendID(a.GetJobId(), a.GetAttempt()),
226 DatabaseID: strconv.FormatInt(a.GetId(), 10),
227 Name: a.GetName(),
228 Size: strconv.FormatInt(a.GetSize(), 10),
229 Digest: a.GetDigest(),
230 }
231 if a.GetCreatedAt() != nil {
232 it.CreatedAt = a.GetCreatedAt().AsTime().UTC().Format(time.RFC3339Nano)
233 }
234 items = append(items, it)
235 }
236 return map[string]any{"artifacts": items}, nil
237}
238
239// signedArtifactURL answers GetSignedArtifactURL: the download URL of the
240// artifact of that name of the run. The body's job backend id is the
241// uploader's (download-artifact copies it from the list), so it is not read.
242func (s *Service) signedArtifactURL(job Job, r *http.Request, req *twirpRequest, expiry time.Time) (any, error) {
243 if req.Name == "" {
244 return nil, &twirpError{http.StatusBadRequest, "invalid_argument", "name is required"}
245 }
246 res, err := job.Client.ListJobArtifacts(r.Context(), connect.NewRequest(&ermite.ListJobArtifactsRequest{JobId: job.ID, Name: req.Name}))
247 if err != nil {
248 return nil, err
249 }
250 var id int64
251 for _, a := range res.Msg.GetArtifacts() {
252 id = max(id, a.GetId())
253 }
254 if id == 0 {
255 return nil, &twirpError{http.StatusNotFound, "not_found", fmt.Sprintf("artifact %q not found", req.Name)}
256 }
257 return map[string]any{"signedUrl": s.blobURL(job, r, opDownload, id, expiry)}, nil
258}
259
260func (s *Service) deleteArtifact(job Job, r *http.Request, req *twirpRequest, _ time.Time) (any, error) {
261 res, err := job.Client.DeleteJobArtifact(r.Context(), connect.NewRequest(&ermite.DeleteJobArtifactRequest{JobId: job.ID, Name: req.Name}))
262 if err != nil {
263 return nil, err
264 }
265 return map[string]any{"ok": true, "artifactId": strconv.FormatInt(res.Msg.GetArtifactId(), 10)}, nil
266}
267
268// twirpFromConnect maps an error of ermite to a Twirp error. ermite's message
269// is kept: it reaches the job log, and it never holds the job token.
270// resource_exhausted is a 412 and not Twirp's 429: the toolkit would retry a
271// 429 five times, and a full quota is not transient.
272func twirpFromConnect(err error) *twirpError {
273 msg := err.Error()
274 var ce *connect.Error
275 if errors.As(err, &ce) {
276 msg = ce.Message()
277 }
278 te := &twirpError{msg: msg}
279 switch connect.CodeOf(err) {
280 case connect.CodeInvalidArgument:
281 te.status, te.code = http.StatusBadRequest, "invalid_argument"
282 case connect.CodeNotFound:
283 te.status, te.code = http.StatusNotFound, "not_found"
284 case connect.CodeAlreadyExists:
285 te.status, te.code = http.StatusConflict, "already_exists"
286 if te.msg == "" {
287 te.msg = msgExists
288 }
289 case connect.CodeFailedPrecondition, connect.CodeResourceExhausted:
290 te.status, te.code = http.StatusPreconditionFailed, "failed_precondition"
291 case connect.CodeUnauthenticated, connect.CodePermissionDenied:
292 te.status, te.code = http.StatusUnauthorized, "unauthenticated"
293 default:
294 te.status, te.code = http.StatusServiceUnavailable, "unavailable"
295 }
296 if te.msg == "" {
297 te.msg = te.code
298 }
299 return te
300}
301
302func writeTwirpError(w http.ResponseWriter, e *twirpError) {
303 var b bytes.Buffer
304 _ = json.NewEncoder(&b).Encode(map[string]string{"code": e.code, "msg": e.msg})
305 w.Header().Set("Content-Type", "application/json")
306 w.Header().Set("Content-Length", strconv.Itoa(b.Len()))
307 w.WriteHeader(e.status)
308 _, _ = w.Write(b.Bytes())
309}
310
311// blobURL is a signed URL of the blob of an artifact, on the host the sandbox
312// used to reach the service.
313func (s *Service) blobURL(job Job, r *http.Request, op string, id int64, expiry time.Time) string {
314 se := expiry.Unix()
315 q := url.Values{"se": {strconv.FormatInt(se, 10)}, "sig": {s.blobSig(job, op, id, se)}}
316 return "http://" + r.Host + BlobPath + op + "/" + strconv.FormatInt(id, 10) + "?" + q.Encode()
317}