Sign in

Search issue titles and bodies, change requests and comments. Private repositories you cannot read are not included.
vole
Clone

Read only, no credentials:

https://rodent.software/vole.git

Write over SSH with a registered key:

git clone ssh://git@rodent.software:2222/vole.git

A free and open-source forge that lives in your Git repo.

Code

395 commit(s) on main

Compare History
update readme Keenan Nemetz · 25m ago 32290114
history blame raw 26.4 KiB
1package ssh
2
3import (
4 "encoding/json"
5 "errors"
6 "fmt"
7 "io"
8 "time"
9
10 xssh "golang.org/x/crypto/ssh"
11 "rodent.software/vole/ci/runs"
12 "rodent.software/vole/core/id"
13 "rodent.software/vole/core/lfs"
14)
15
16// The runner protocol.
17//
18// A runner connects over the same SSH transport as everything else, so it
19// authenticates with a key and needs no token: the account it connects as decides
20// which repositories it may run jobs for. One JSON object per line, one reply per
21// request, on a single channel.
22//
23// {"op":"lease"} -> {"run":{...}} or {}
24// {"op":"heartbeat","run":"<id>"} -> {"ok":true|false}
25// {"op":"finish","run":"<id>","record":{...}} -> {"ok":true|false}
26//
27// A finish carries the attempt's record: the runner uploads its log and artifacts,
28// then sends the record on this channel. The server checks that it names the
29// account that holds the lease, that it describes the job the queue handed out, and
30// that every log and artifact pointer exists in the object store. It then stores the
31// attempt and the run's result in one transaction. The record is not a git commit;
32// CI history is instance-local.
33//
34// A tagged build that publishes release artifacts still pushes the release pointers
35// to the metadata branch after the record is stored. Releases are product metadata,
36// not CI history.
37//
38// "ok":false means the lease is gone — another runner has the job — and the runner
39// must stop rather than report a result nobody will record. A retry of the same
40// record after a lost reply is answered "ok":true, because the bytes are already
41// stored and the attempt is append-only.
42
43const (
44 // eligibleFor is how long a session reuses its repository list. Access rules live
45 // in the instance database and only change when an admin edits them, so a runner
46 // re-derives its list on the roster's five-second beat instead of on every poll.
47 eligibleFor = 5 * time.Second
48 // maxLeaseWait caps what a runner may ask for. The wait is a hint — the lease
49 // remains the database's decision — so all a long wait holds is one SSH channel.
50 maxLeaseWait = 60 * time.Second
51 // maxRunnerRequestBytes bounds one request line. A lease is small, but a finish
52 // carries the attempt's steps and artifact names, so the bound has room for a
53 // run at the artifact cap. It is per line, so a long-lived runner is not limited
54 // in total.
55 maxRunnerRequestBytes = 4 << 20
56)
57
58// errRequestTooLarge ends a runner session whose request line outgrew the bound.
59var errRequestTooLarge = errors.New("runner request line is too large")
60
61// lineLimitReader fails once a single newline-terminated line passes limit bytes.
62// It counts bytes as they cross the reader, so json.Decoder's internal buffering
63// cannot hide a large request, and it resets at every newline so one oversized
64// line is the only thing it refuses.
65type lineLimitReader struct {
66 r io.Reader
67 limit int64
68 line int64
69}
70
71func (l *lineLimitReader) Read(p []byte) (int, error) {
72 n, err := l.r.Read(p)
73 for _, b := range p[:n] {
74 if b == '\n' {
75 l.line = 0
76 continue
77 }
78 l.line++
79 if l.line > l.limit {
80 return 0, errRequestTooLarge
81 }
82 }
83 return n, err
84}
85
86type runnerRequest struct {
87 Op string `json:"op"`
88 Run string `json:"run"`
89 // Repos narrows this session's leases to the repositories the runner was started
90 // for (--repos). It is applied as an intersection with what the account's runner
91 // allowlists grant, so it can only subtract: a runner cannot ask its way into a
92 // repository no allowlist names. Empty means "no narrowing", which keeps the list
93 // following allowlist edits instead of freezing it at connect time.
94 Repos []string `json:"repos,omitempty"`
95 // Labels are what this machine is (--label), and a job is only leased to a runner
96 // carrying every label the job asked for with runs-on. They describe a machine, not
97 // a permission: a runner that claims a label it does not have only hurts itself,
98 // which is why this is a client claim where --repos is server-enforced.
99 Labels []string `json:"labels,omitempty"`
100 // Network is the network this machine can give, derived from --container-network.
101 // A job that declares `network: bridge` is only leased to a runner that says it can
102 // provide one, so the job waits for a capable machine instead of running without
103 // the network it asked for. A runner that claims more than it can give only fails
104 // its own jobs, which is the same trade as a label.
105 Network string `json:"network,omitempty"`
106 // Wait is how many seconds an idle lease may park before answering "nothing to
107 // do". Zero answers at once, which is what a one-shot runner wants: it is here to
108 // run the job it was started for, not to wait for the next one.
109 Wait int `json:"wait,omitempty"`
110 // Record is the attempt's result, sent with finish. The runner no longer
111 // publishes it to Git: this channel is where the server receives it, checks it
112 // against the lease, and stores it.
113 Record *runs.Record `json:"record,omitempty"`
114 // Log is one chunk of the attempt's log, sent while the job runs.
115 Log *runnerLogChunk `json:"log,omitempty"`
116}
117
118// runnerLogChunk is one piece of an attempt's log. Offset is where it starts in
119// the attempt's file, so the server can refuse a gap or a duplicate.
120type runnerLogChunk struct {
121 Attempt int `json:"attempt"`
122 Offset int64 `json:"offset"`
123 Data []byte `json:"data"`
124}
125
126// runnerCredential is the private half of a run's key. It is delivered once, written
127// into the job's run directory, and never stored by the instance.
128type runnerCredential struct {
129 PrivateKey string `json:"private_key"`
130 Fingerprint string `json:"fingerprint"`
131}
132
133// runnerSecret is one delivered value and the version it came from, so the attempt
134// record can say which value ran without carrying it.
135type runnerSecret struct {
136 Version int `json:"version"`
137 Value string `json:"value"`
138}
139
140// runnerSecretIssue is a declared secret the repository could not supply. Reason is
141// empty when the reason is simply that nobody has stored that name.
142type runnerSecretIssue struct {
143 Name string `json:"name"`
144 Reason string `json:"reason,omitempty"`
145}
146
147type runnerStep struct {
148 Name string `json:"name"`
149 Run string `json:"run"`
150 Timeout string `json:"timeout"`
151}
152
153type runnerService struct {
154 ID string `json:"id"`
155 Image string `json:"image"`
156 Env map[string]string `json:"env,omitempty"`
157}
158
159type runnerJob struct {
160 ID string `json:"id"`
161 Repo string `json:"repo"`
162 SHA string `json:"sha"`
163 // Attempt is which try at this job this is. The record lives under it, so the
164 // runner has to know the number it is reporting.
165 Attempt int `json:"attempt"`
166 // Ref, WorkflowSHA and StepsSHA are what the runner signs when the attempt ends:
167 // the exact revision, the exact workflow file, and the digest of the steps that
168 // were pinned when the job was queued. Without them the record could only say
169 // "some job", which is not a statement anybody can check later.
170 Ref string `json:"ref"`
171 WorkflowSHA string `json:"workflow_sha"`
172 StepsSHA string `json:"steps_sha"`
173 Branch string `json:"branch"`
174 Workflow string `json:"workflow"`
175 Job string `json:"job"`
176 Steps []runnerStep `json:"steps"`
177 // Trigger and EventPayload are the pinned event that requested this run. The
178 // runner writes the payload into a file and sets VOLE_EVENT, VOLE_EVENT_ACTION
179 // and VOLE_EVENT_PATH; it is data, never an interpolated command.
180 Trigger string `json:"trigger,omitempty"`
181 TriggerAction string `json:"trigger_action,omitempty"`
182 EventPayload string `json:"event_payload,omitempty"`
183 // Network is the isolation the job asked for with `network:`, or empty when the
184 // workflow did not say. The runner applies it to the container it starts, so a job
185 // that asks for none gets none even on a machine that could provide a bridge.
186 Network string `json:"network,omitempty"`
187 // Bot is the account this job acts as, and Credential is the key it may do it with.
188 // The private half is here and nowhere else: the queue holds the public half and
189 // the account name, so a credential can be refused the moment the run ends.
190 Bot string `json:"bot,omitempty"`
191 Credential *runnerCredential `json:"credential,omitempty"`
192 // CredentialError says the job asked to act as an account it cannot have. The
193 // runner fails the job with this line rather than running it without the key,
194 // because a step that cannot write is a broken job, not a degraded one.
195 CredentialError string `json:"credential_error,omitempty"`
196 // HostKey is where to find the instance when the job pushes, paired with the
197 // credential above.
198 HostKey string `json:"host_key,omitempty"`
199 // Timeout is the job's budget as a duration string, or empty when it has none.
200 Timeout string `json:"timeout,omitempty"`
201 // Image is a registry reference or "file:<name>". For the latter, ImageDefinition
202 // is the pinned recipe and ImageSHA is its digest; the runner builds it locally.
203 Image string `json:"image"`
204 ImageFile string `json:"image_file,omitempty"`
205 ImageDefinition string `json:"image_definition,omitempty"`
206 ImageSHA string `json:"image_sha,omitempty"`
207 // Env is the job's declared environment, pinned when the run was queued.
208 Env map[string]string `json:"env,omitempty"`
209 // Services are the containers the job asked to run beside it, pinned when the
210 // run was queued. The runner gives the job and these one network for the attempt.
211 Services []runnerService `json:"services,omitempty"`
212 // Secrets are the values the job declared that this repository has, and Missing
213 // are the names it declared that it does not. A missing name is not a failure:
214 // the variable is left unset and the runner says so at the top of the log. The
215 // values travel here and nowhere else — the queue row holds the names only.
216 Secrets map[string]runnerSecret `json:"secrets,omitempty"`
217 Missing []runnerSecretIssue `json:"missing_secrets,omitempty"`
218 // Artifacts are the declared patterns the runner expands and uploads after the
219 // steps succeed.
220 Artifacts []string `json:"artifacts,omitempty"`
221 // Tag is set for a tag-triggered run: the release being built. TagAnnotated says
222 // whether that tag carries notes, which is what makes it a release.
223 Tag string `json:"tag,omitempty"`
224 TagAnnotated bool `json:"tag_annotated,omitempty"`
225 // ReleaseArtifacts are the patterns that also publish to the tag's release.
226 ReleaseArtifacts []string `json:"release_artifacts,omitempty"`
227 // Pages is the site deploy this run performs, if any. The branch comes from
228 // instance configuration, which is why it is in the payload rather than in the
229 // ci.
230 Pages *runnerPages `json:"pages,omitempty"`
231}
232
233type runnerPages struct {
234 Path string `json:"path"`
235 Ref string `json:"ref"`
236}
237
238type runnerReply struct {
239 Op string `json:"op"`
240 Error string `json:"error,omitempty"`
241 Run *runnerJob `json:"run,omitempty"`
242 OK bool `json:"ok"`
243 Lease int `json:"lease,omitempty"`
244 // Who am I: the account name this channel authenticated as. A runner answers
245 // "ping" with this so it can say whose results it will be reporting, and so a
246 // --repos filter can be checked against what the account may run.
247 Name string `json:"name,omitempty"`
248 // Repos are the repositories this account may run jobs for, sent with "ping" so
249 // a runner started with --repos can find out at startup that its filter matches
250 // nothing — a typo it would otherwise discover as a permanently idle runner —
251 // instead of after the first build that never arrived.
252 Repos []string `json:"repos,omitempty"`
253 // LogSize is how much of the log the server has after a log chunk.
254 LogSize int64 `json:"log_size,omitempty"`
255 // Cancel names the account that asked this running job to stop. It rides the
256 // heartbeat reply, so stopping a run needs no new traffic and a runner too old to
257 // know the field ignores it. Empty means nobody has asked.
258 Cancel string `json:"cancel,omitempty"`
259}
260
261// serveRunner answers one runner session until it disconnects.
262func (s *Server) serveRunner(ch xssh.Channel, user string) {
263 defer ch.Close()
264 cfg := s.cfg
265 if cfg.Runs == nil {
266 _ = json.NewEncoder(ch).Encode(runnerReply{Op: "error", Error: "the run queue is not configured on this server"})
267 return
268 }
269 dec := json.NewDecoder(&lineLimitReader{r: ch, limit: maxRunnerRequestBytes})
270 enc := json.NewEncoder(ch)
271 owner := user
272 // Presence is one row per channel: a pool of three runners on one account is
273 // three rows, and the row goes away when the channel does. Nothing else records
274 // that a runner exists, so this is what the runners page shows.
275 sessionID, err := id.New(time.Now().UTC())
276 if err != nil {
277 _ = enc.Encode(runnerReply{Op: "error", Error: err.Error()})
278 return
279 }
280 if err := cfg.Runs.TouchSession(sessionID, user, nil, time.Now().UTC()); err != nil {
281 cfg.Log.Warn("runner: cannot record the session", "user", user, "err", err)
282 }
283 defer cfg.Runs.DropSession(sessionID)
284 // repos is this session's eligibility list, reused for one poll window. Reading
285 // every repository's rules on every lease request is O(repos) per idle runner, and
286 // the answer changes only when an admin edits an allowlist — so it is refreshed on
287 // the same five-second beat the roster is, and not once per request.
288 var repos []string
289 var reposAt time.Time
290 cfg.Log.Info("runner connected", "user", user)
291
292 for {
293 var req runnerRequest
294 if err := dec.Decode(&req); err != nil {
295 if errors.Is(err, errRequestTooLarge) {
296 _ = enc.Encode(runnerReply{Op: "error", Error: fmt.Sprintf(
297 "a runner request must fit in %d bytes", maxRunnerRequestBytes)})
298 cfg.Log.Warn("runner: refused an oversized request", "user", user)
299 }
300 return // disconnected, or refused
301 }
302 switch req.Op {
303 case "lease":
304 now := time.Now().UTC()
305 // A runner may only run jobs for repositories whose runner allowlist it is
306 // on: running a job executes that repository's code, which is a bigger
307 // power than writing to it.
308 if reposAt.IsZero() || now.Sub(reposAt) > eligibleFor {
309 list, err := s.runnableRepos(user)
310 if err != nil {
311 _ = enc.Encode(runnerReply{Op: "lease", Error: err.Error()})
312 continue
313 }
314 repos, reposAt = list, now
315 }
316 // --repos narrows the grant. The intersection is computed here rather than
317 // trusted from the runner, so the worst a hostile client can do is ask for
318 // nothing — never for a repository no allowlist gave it.
319 want := narrow(repos, req.Repos)
320 // Take the wake-up before asking, so a run queued between the lease and the
321 // wait closes a channel somebody is holding. Asking first and waiting second
322 // loses exactly that race.
323 wake := cfg.Runs.Work()
324 lease := runs.LeaseRequest{
325 Owner: owner, Repos: want, Labels: req.Labels, Network: req.Network,
326 TTL: cfg.Lease, MaxPerRepo: cfg.MaxJobsPerRepo,
327 }
328 run, err := cfg.Runs.Lease(lease, now)
329 if err != nil {
330 _ = enc.Encode(runnerReply{Op: "lease", Error: err.Error()})
331 continue
332 }
333 if run == nil && req.Wait > 0 {
334 wait := time.Duration(req.Wait) * time.Second
335 if wait > maxLeaseWait {
336 wait = maxLeaseWait
337 }
338 select {
339 case <-wake:
340 case <-time.After(wait):
341 case <-s.done:
342 }
343 // Ask again rather than assume: the wake-up may belong to a job another
344 // runner took first, and a lease is the only thing that hands work out.
345 run, err = cfg.Runs.Lease(lease, time.Now().UTC())
346 if err != nil {
347 _ = enc.Encode(runnerReply{Op: "lease", Error: err.Error()})
348 continue
349 }
350 }
351 if run == nil {
352 // An idle runner is still connected: refresh its presence so the page can
353 // say "waiting for work" instead of showing a runner that vanished. The
354 // labels ride along, because a runner waiting for the wrong kind of work
355 // is exactly what this row is for.
356 _ = cfg.Runs.TouchSession(sessionID, user, req.Labels, time.Now().UTC())
357 _ = enc.Encode(runnerReply{Op: "lease", OK: true})
358 continue
359 }
360 // The values are resolved here, at the moment the job is handed out, so
361 // the queue never holds one and a rotation between queueing and leasing
362 // delivers the value the job actually ran with.
363 secrets, missing := s.leaseSecrets(run, user)
364 job := &runnerJob{
365 ID: run.ID, Repo: run.Repo, SHA: run.SHA, Attempt: run.Attempts, Branch: run.Branch,
366 Ref: run.Ref, Workflow: run.Workflow, WorkflowSHA: run.WorkflowSHA,
367 StepsSHA: run.StepsSHA, Job: run.Job, Artifacts: run.Artifacts,
368 Trigger: run.Trigger, TriggerAction: run.TriggerAction, EventPayload: run.EventPayload,
369 Image: run.Image, ImageFile: run.ImageFile, ImageDefinition: run.ImageDefinition,
370 ImageSHA: run.ImageSHA,
371 Network: run.Network,
372 Timeout: run.Timeout,
373 Env: run.Env,
374 Secrets: secrets, Missing: missing,
375 Tag: run.Tag, TagAnnotated: run.TagAnnotated, ReleaseArtifacts: run.ReleaseArtifacts,
376 }
377 if run.PagesPath != "" {
378 ref := cfg.PagesRef
379 if ref == "" {
380 ref = "pages"
381 }
382 job.Pages = &runnerPages{Path: run.PagesPath, Ref: ref}
383 }
384 for _, st := range run.Steps {
385 job.Steps = append(job.Steps, runnerStep{Name: st.Name, Run: st.Run, Timeout: st.Timeout})
386 }
387 for _, svc := range run.Services {
388 job.Services = append(job.Services, runnerService{ID: svc.ID, Image: svc.Image, Env: svc.Env})
389 }
390 // A job that acts as a bot gets a key for this run only. The account and
391 // the repository are pinned in the row, so the hook can refuse a write that
392 // the account's role would not allow and the key stops working when the run
393 // ends. A bot that cannot be used fails the job here rather than at its
394 // first push, which is a much better place to learn it.
395 if run.Bot != "" {
396 cred, hostKey, reason := s.mintCredential(run, now)
397 if reason != "" {
398 job.CredentialError = reason
399 } else {
400 job.Bot, job.Credential, job.HostKey = run.Bot, cred, hostKey
401 }
402 }
403 cfg.Log.Info("runner leased a job", "user", user, "repo", run.Repo,
404 "job", run.Job, "sha", run.SHA[:8], "run", run.ID, "attempt", run.Attempts)
405 _ = cfg.Runs.SessionWorking(sessionID, run.Repo, run.ID, time.Now().UTC())
406 _ = enc.Encode(runnerReply{Op: "lease", OK: true, Run: job, Lease: int(cfg.Lease.Seconds())})
407
408 case "log":
409 if req.Log == nil {
410 _ = enc.Encode(runnerReply{Op: "log", Error: "a log request carries no chunk"})
411 continue
412 }
413 run, err := cfg.Runs.Get(req.Run)
414 if err != nil {
415 _ = enc.Encode(runnerReply{Op: "log", Error: err.Error()})
416 continue
417 }
418 if run == nil {
419 _ = enc.Encode(runnerReply{Op: "log", Error: fmt.Sprintf("no run %s", req.Run)})
420 continue
421 }
422 if req.Log.Attempt != run.Attempts {
423 _ = enc.Encode(runnerReply{Op: "log", Error: fmt.Sprintf(
424 "%s: the log is attempt %d but this runner holds attempt %d",
425 req.Run, req.Log.Attempt, run.Attempts)})
426 continue
427 }
428 held, err := cfg.Runs.Held(run.ID, owner)
429 if err != nil {
430 _ = enc.Encode(runnerReply{Op: "log", Error: err.Error()})
431 continue
432 }
433 if !held {
434 _ = enc.Encode(runnerReply{Op: "log", OK: false})
435 continue
436 }
437 if cfg.Logs == nil {
438 _ = enc.Encode(runnerReply{Op: "log", Error: "this instance has no log directory"})
439 continue
440 }
441 size, err := cfg.Logs.Append(run.Repo, run.ID, req.Log.Attempt, req.Log.Offset, req.Log.Data)
442 if err != nil {
443 _ = enc.Encode(runnerReply{Op: "log", Error: err.Error()})
444 continue
445 }
446 if ok, err := cfg.Runs.RecordLogSize(run.ID, owner, req.Log.Attempt, size, time.Now().UTC()); err != nil {
447 _ = enc.Encode(runnerReply{Op: "log", Error: err.Error()})
448 continue
449 } else if !ok {
450 _ = enc.Encode(runnerReply{Op: "log", OK: false})
451 continue
452 }
453 _ = enc.Encode(runnerReply{Op: "log", OK: true, LogSize: size})
454
455 case "heartbeat":
456 // Liveness only: the lease is what a heartbeat buys. Presence is refreshed
457 // here too, so a long build does not make its runner look gone. A stop request
458 // rides the same reply rather than a new message: the runner already asks this
459 // question every few seconds, and a runner that does not know the field keeps
460 // working exactly as before.
461 _ = cfg.Runs.TouchSession(sessionID, user, nil, time.Now().UTC())
462 held, cancelBy, err := cfg.Runs.Heartbeat(req.Run, owner, cfg.Lease, time.Now().UTC())
463 if err != nil {
464 _ = enc.Encode(runnerReply{Op: "heartbeat", Error: err.Error()})
465 continue
466 }
467 _ = enc.Encode(runnerReply{Op: "heartbeat", OK: held, Cancel: cancelBy})
468
469 case "finish":
470 // The result arrives here, not on the metadata branch. The record is
471 // checked against the lease and the job it claims to describe, then
472 // stored with the run's status in one transaction.
473 _ = cfg.Runs.SessionIdle(sessionID, time.Now().UTC())
474 run, err := cfg.Runs.Get(req.Run)
475 if err != nil {
476 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
477 continue
478 }
479 if run == nil {
480 _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf("no run %s", req.Run)})
481 continue
482 }
483 if req.Record == nil {
484 _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf(
485 "%s: this runner did not send a record; upgrade the runner and the server together", req.Run)})
486 continue
487 }
488 if req.Record.Attempt != run.Attempts {
489 _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf(
490 "%s: the record is attempt %d but this runner holds attempt %d",
491 req.Run, req.Record.Attempt, run.Attempts)})
492 continue
493 }
494 if req.Record.Signer != owner {
495 _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf(
496 "%s: the record names %s as the reporter, but %s reported it",
497 req.Run, req.Record.Signer, owner)})
498 continue
499 }
500 if err := checkRecordPointers(cfg.LFS, run.Repo, *req.Record); err != nil {
501 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
502 continue
503 }
504 if req.Record.LogSize == 0 {
505 // A failed stream leaves a partial file, and an attempt that produced
506 // no output leaves none. Either way the result names no log.
507 if cfg.Logs != nil {
508 if err := cfg.Logs.Remove(run.Repo, run.ID, req.Record.Attempt); err != nil {
509 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
510 continue
511 }
512 }
513 } else {
514 if cfg.Logs == nil {
515 _ = enc.Encode(runnerReply{Op: "finish", Error: "this instance has no log directory"})
516 continue
517 }
518 size, err := cfg.Logs.Size(run.Repo, run.ID, req.Record.Attempt)
519 if err != nil {
520 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
521 continue
522 }
523 if size != req.Record.LogSize {
524 _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf(
525 "%s: the log is %d bytes but the record says %d",
526 req.Run, size, req.Record.LogSize)})
527 continue
528 }
529 if err := cfg.Logs.Sync(run.Repo, run.ID, req.Record.Attempt); err != nil {
530 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
531 continue
532 }
533 }
534 ok, err := cfg.Runs.RecordAttempt(run.ID, owner, *req.Record, time.Now().UTC())
535 if err != nil {
536 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
537 continue
538 }
539 // The run is over, so the key it could act with is over too. The expiry would
540 // catch it eventually; this is the part that does not wait.
541 if err := cfg.Runs.DropRunCredentials(run.ID); err != nil {
542 cfg.Log.Warn("runner: cannot drop the run credential", "run", run.ID, "err", err)
543 }
544 cfg.Log.Info("run finished", "user", user, "run", run.ID, "exit", req.Record.Exit,
545 "signer", req.Record.Signer, "recorded", ok)
546 _ = enc.Encode(runnerReply{Op: "finish", OK: ok})
547
548 case "ping":
549 reply := runnerReply{Op: "ping", OK: true, Name: user}
550 // What this account may run, so a runner started with --repos can say at
551 // startup that its filter matches nothing rather than idling in silence.
552 if granted, gerr := s.runnableRepos(user); gerr == nil {
553 reply.Repos = granted
554 } else {
555 cfg.Log.Warn("runner: cannot list runnable repositories", "user", user, "err", gerr)
556 }
557 _ = enc.Encode(reply)
558
559 default:
560 _ = enc.Encode(runnerReply{Op: req.Op, Error: "unknown op"})
561 }
562 }
563}
564
565// checkRecordPointers refuses a record whose log or artifacts are not actually in
566// the object store. A push used to be checked for pointer hygiene at the boundary;
567// the finish channel is the boundary now.
568func checkRecordPointers(objects *lfs.Store, repoName string, rec runs.Record) error {
569 for _, a := range rec.Artifacts {
570 if err := checkPointer(objects, repoName, lfs.Pointer{OID: a.OID, Size: a.Size}); err != nil {
571 return fmt.Errorf("artifact %s: %w", a.Name, err)
572 }
573 }
574 return nil
575}
576
577func checkPointer(objects *lfs.Store, repoName string, p lfs.Pointer) error {
578 if objects == nil {
579 return fmt.Errorf("this instance has no object store for run outputs")
580 }
581 size, err := objects.Stat(repoName, p.OID)
582 if err != nil {
583 return fmt.Errorf("object %s is missing: %w", p.OID, err)
584 }
585 if size != p.Size {
586 return fmt.Errorf("object %s is %d bytes, pointer says %d", p.OID, size, p.Size)
587 }
588 return nil
589}
590
591// runnableRepos lists the repositories this account may execute jobs for.
592func (s *Server) runnableRepos(name string) ([]string, error) {
593 repos, err := s.cfg.Repos.List()
594 if err != nil {
595 return nil, err
596 }
597 var out []string
598 for _, m := range repos {
599 a, err := m.ACL()
600 if err != nil {
601 continue // a repository without readable rules grants nothing
602 }
603 // Whose results count is the same list as whose job may be leased: running a
604 // job executes that repository's code, and a result from an account the
605 // repository will not count is only worth refusing here rather than after
606 // the build.
607 if a.CanReport(name) {
608 out = append(out, m.Name)
609 }
610 }
611 return out, nil
612}
613
614// narrow keeps only the repositories a runner asked for, and is the whole of what
615// --repos is allowed to do. The granted list is what the account's runner allowlists
616// produce; the asked list comes from the client. Intersecting them means the worst a
617// client can do is run less than it was allowed to — the alternative, honouring the
618// client's list, would make an SSH flag an access-control decision.
619//
620// An empty asked list means "no narrowing" rather than "nothing": that is what a runner
621// without --repos sends, and it keeps such a runner following allowlist edits instead
622// of freezing its list at connect time. Asked names that were never granted simply
623// disappear, which the runner reports on startup because it can see both lists.
624func narrow(granted, asked []string) []string {
625 if len(asked) == 0 {
626 return granted
627 }
628 want := make(map[string]bool, len(asked))
629 for _, r := range asked {
630 want[r] = true
631 }
632 out := make([]string, 0, len(granted))
633 for _, r := range granted {
634 if want[r] {
635 out = append(out, r)
636 }
637 }
638 return out
639}
640