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/git/vole.git

Write over SSH with a registered key:

git clone ssh://git@0.0.0.0:2222/vole.git

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

Phase 4: the runner publishes its result as a signed commit

90fdaf397a1120677a1246201369e6dcc82b0aed · Keenan Nemetz · 16d ago · parent 96cbb6fa

Phase 4: the runner publishes its result as a signed commit

A finished run's result existed only in data/runs.db and the object store, so a
clone carried the commits and the workflow but not the fact that CI passed, who
said so, or the log it said it with. And the previous cut of this change had the
server commit the record — which is the box vouching for a party, the same
mistake as the instance signing key that was deleted.

The runner now publishes it. When an attempt ends the runner commits
runs/<run id>/<attempt>/ — record.md (repo, ref, sha, workflow blob, job, steps
digest, exit, finish time, signer, key) plus an LFS pointer for the log and one
per artifact — together with any release pointers the build published, and
pushes that once. One push per run: the result and the artifacts it produced are
one fact, so a release cannot carry bytes whose build has no record.

The commit is signed with the runner's SSH key, and that signature is the whole
attestation: it covers the tree, the parents and the message, so the separate
claim and signature file are gone, and there is no second statement to keep in
step with the files it describes. pre-receive checks it against an
allowed-signers file built from the pushing account's keys, so identity comes
from the key and not from a name typed into a commit.

Who may report is the repository's runner allowlist — the same list that decides
who may lease a job — and the record must describe the job the queue handed that
account: same run, same attempt, same commit, workflow, job and steps digest.
Without that an allowed runner could report a green result for a commit nobody
asked it to build, or invent attempt 99 and have the next rebuild believe it.
Records are append-only, with no admin override: an edit or a deletion is a
rewrite of evidence, and the alternative to refusing is running the job again.

runs.db keeps the queue (queued and leased jobs have no record because they have
not happened) and mirrors finished runs as derived rows, which vole index
rebuild reconstructs from the branch. The attestation row now names the commit
instead of storing a claim and a signature.

Measured: a real run finished, a bare clone carried runs/<id>/1/record.md, git
verify-commit accepted that commit under the roster and refused an unsigned one;
a writer who is not a runner was refused by pre-receive naming the allowlist;
one CI release commit carried both the record and releases/v1.0.0/tool-linux;
deleting runs.db and rebuilding restored the row, the attestation and the log.

internal/sshserver/runner.go +68−194

@@ -1,20 +1,15 @@
1 1 package sshserver
2 2
3 3 import (
4 "crypto/ed25519"
5 4 "encoding/json"
6 5 "fmt"
7 6 "time"
8 7
9 "github.com/rodent-software/vole/internal/lfs"
8 "github.com/rodent-software/vole/internal/issues"
10 9 "github.com/rodent-software/vole/internal/runs"
11 10 "golang.org/x/crypto/ssh"
12 11 )
13 12
14 // maxRunArtifacts bounds one run's output list. It is a page of downloads, not a
15 // package index.
16 const maxRunArtifacts = 64
17
18 13 // The runner protocol.
19 14 //
20 15 // A runner connects over the same SSH transport as everything else, so it
@@ -22,45 +17,25 @@ const maxRunArtifacts = 64
22 17 // which repositories it may run jobs for. One JSON object per line, one reply per
23 18 // request, on a single channel.
24 19 //
25 // {"op":"lease"} -> {"run":{...}} or {}
26 // {"op":"heartbeat","run":"<id>"} -> {"ok":true|false}
27 // {"op":"finish","run":"<id>","exit":0,
28 // "log_oid":"<sha256>","log_size":1234} -> {"ok":true|false}
20 // {"op":"lease"} -> {"run":{...}} or {}
21 // {"op":"heartbeat","run":"<id>"} -> {"ok":true|false}
22 // {"op":"finish","run":"<id>"} -> {"ok":true|false}
23 //
24 // A finish carries no result: the result is on the branch already. The runner
25 // uploads its log and artifacts, commits the attempt's record under
26 // `runs/<run id>/<attempt>/` together with any release pointers the build published,
27 // pushes that once, and only then says it is done. `pre-receive` is what accepts or
28 // refuses it, which is where the runner allowlist and the signature are checked; this
29 // channel can only close the lease and mirror what the push already established.
30 // Nothing here writes a result, so there is nothing here for the instance to vouch
31 // for.
29 32 //
30 33 // "ok":false means the lease is gone — another runner has the job — and the runner
31 34 // must stop rather than report a result nobody will record.
32 //
33 // A log is not streamed. The runner uploads its own bytes through the same
34 // git-lfs-transfer put-object path it uses for artifacts, and names the object here:
35 // the id is computed from the content by the party that produced it, so the
36 // archived log is something that party can attest to rather than a record the
37 // server assembled.
38 35
39 36 type runnerRequest struct {
40 Op string `json:"op"`
41 Run string `json:"run"`
42 Exit int `json:"exit"`
43 // LogOID names the log object the runner uploaded for this attempt, and
44 // LogSize is how many bytes it holds. Both are absent when a log could not be
45 // stored: a result is worth recording even when its output is not.
46 LogOID string `json:"log_oid,omitempty"`
47 LogSize int64 `json:"log_size,omitempty"`
48 // Artifacts arrive with the finish, not as a separate call: a job's outputs
49 // and its result are one fact, so there is no window in which a successful run
50 // has no artifacts or a failed one has some.
51 Artifacts []runnerArtifact `json:"artifacts,omitempty"`
52 // Finished is the runner's own unix time for when the attempt ended, and
53 // Signature is its signature over the claim that includes it. A finish without
54 // a signature is refused: an unattributed result is what the queue already had,
55 // and the runner is the only party that can be held to it.
56 Finished int64 `json:"finished,omitempty"`
57 Signature []byte `json:"signature,omitempty"`
58 }
59
60 type runnerArtifact struct {
61 Name string `json:"name"`
62 OID string `json:"oid"`
63 Size int64 `json:"size"`
37 Op string `json:"op"`
38 Run string `json:"run"`
64 39 }
65 40
66 41 type runnerStep struct {
@@ -73,9 +48,12 @@ type runnerJob struct {
73 48 ID string `json:"id"`
74 49 Repo string `json:"repo"`
75 50 SHA string `json:"sha"`
51 // Attempt is which try at this job this is. The record lives under it, so the
52 // runner has to know the number it is reporting.
53 Attempt int `json:"attempt"`
76 54 // Ref, WorkflowSHA and StepsSHA are what the runner signs when the attempt ends:
77 55 // the exact revision, the exact workflow file, and the digest of the steps that
78 // were pinned when the job was queued. Without them the claim could only say
56 // were pinned when the job was queued. Without them the record could only say
79 57 // "some job", which is not a statement anybody can check later.
80 58 Ref string `json:"ref"`
81 59 WorkflowSHA string `json:"workflow_sha"`
@@ -157,7 +135,7 @@ func (s *Server) serveRunner(ch ssh.Channel, user, accountID string) {
157 135 continue
158 136 }
159 137 job := &runnerJob{
160 ID: run.ID, Repo: run.Repo, SHA: run.SHA, Branch: run.Branch,
138 ID: run.ID, Repo: run.Repo, SHA: run.SHA, Attempt: run.Attempts, Branch: run.Branch,
161 139 Ref: run.Ref, Workflow: run.Workflow, WorkflowSHA: run.WorkflowSHA,
162 140 StepsSHA: run.StepsSHA, Job: run.Job, Artifacts: run.Artifacts,
163 141 Tag: run.Tag, TagAnnotated: run.TagAnnotated, ReleaseArtifacts: run.ReleaseArtifacts,
@@ -186,57 +164,70 @@ func (s *Server) serveRunner(ch ssh.Channel, user, accountID string) {
186 164 _ = enc.Encode(runnerReply{Op: "heartbeat", OK: ok})
187 165
188 166 case "finish":
189 // Everything is checked before anything is recorded. A finish that
190 // cannot be attributed must not leave artifacts, a log or a result
191 // behind: the point of the signature is that the record names the party
192 // that produced it, and half a record is worse than none.
193 //
194 // Artifacts and the log are checked against the store — a runner naming
195 // bytes it never uploaded would otherwise produce a download that 404s —
196 // and the claim is assembled from the row plus what the runner says it
197 // produced, then verified under the account the channel authenticated as.
198 arts, err := s.checkArtifacts(req.Artifacts)
167 // Nothing is created here. The runner's record is already on the branch,
168 // checked by `pre-receive`; this closes the lease and mirrors the record
169 // into the queue so a page or a gate needs no tree walk.
170 run, err := cfg.Runs.Get(req.Run)
199 171 if err != nil {
200 172 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
201 173 continue
202 174 }
203 if req.LogOID != "" {
204 if err := s.checkRunLog(req.LogOID, req.LogSize); err != nil {
205 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
206 continue
207 }
175 if run == nil {
176 _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf("no run %s", req.Run)})
177 continue
208 178 }
209 payload, err := s.claimFor(req, arts)
179 now := time.Now().UTC()
180 // The queue may already have given this job away: a runner whose lease was
181 // taken over is finishing an attempt somebody else owns.
182 held, err := cfg.Runs.Held(run.ID, owner)
210 183 if err != nil {
211 184 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
212 185 continue
213 186 }
214 if len(req.Signature) == 0 {
215 _ = enc.Encode(runnerReply{Op: "finish",
216 Error: "a finish must be signed by the runner that ran the job: run `vole runner --key <private key>`"})
187 if !held {
188 _ = enc.Encode(runnerReply{Op: "finish", OK: false})
217 189 continue
218 190 }
219 fingerprint, err := s.verifyClaim(accountID, payload, req.Signature)
191 m, err := cfg.Repos.Open(run.Repo)
220 192 if err != nil {
221 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
193 _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf("cannot open %s: %v", run.Repo, err)})
222 194 continue
223 195 }
224
225 now := time.Now().UTC()
226 if len(arts) > 0 {
227 ok, err := cfg.Runs.RecordArtifacts(req.Run, owner, arts)
196 rec, err := runs.ReadRecord(m.Git(), issues.Ref, run.ID)
197 if err != nil {
198 _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf("cannot read the record for %s: %v", run.ID, err)})
199 continue
200 }
201 switch {
202 case rec == nil:
203 _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf(
204 "no record for run %s on %s: a runner reports a result by pushing %s, and nothing is recorded without one",
205 run.ID, issues.Ref, runs.RecordPath(run.ID, run.Attempts))})
206 continue
207 case rec.Attempt != run.Attempts:
208 _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf(
209 "the record for %s is attempt %d but this runner holds attempt %d",
210 run.ID, rec.Attempt, run.Attempts)})
211 continue
212 case rec.Signer != accountID:
213 _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf(
214 "the record for %s was reported by another account; %s cannot finish it",
215 run.ID, user)})
216 continue
217 }
218 if len(rec.Artifacts) > 0 {
219 ok, err := cfg.Runs.RecordArtifacts(run.ID, owner, rec.Artifacts)
228 220 if err != nil {
229 221 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
230 222 continue
231 223 }
232 224 if !ok {
233 // The lease moved: report nothing, exactly like a lost finish.
234 225 _ = enc.Encode(runnerReply{Op: "finish", OK: false})
235 226 continue
236 227 }
237 228 }
238 if req.LogOID != "" {
239 recorded, err := cfg.Runs.RecordLog(req.Run, owner, req.LogOID, req.LogSize, now)
229 if rec.Log != nil {
230 recorded, err := cfg.Runs.RecordLog(run.ID, owner, rec.Log.OID, rec.Log.Size, now)
240 231 if err != nil {
241 232 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
242 233 continue
@@ -246,11 +237,10 @@ func (s *Server) serveRunner(ch ssh.Channel, user, accountID string) {
246 237 continue
247 238 }
248 239 }
249 attested, err := cfg.Runs.RecordAttestation(req.Run, owner, runs.Attestation{
250 Signer: accountID,
251 Fingerprint: fingerprint,
252 Payload: payload,
253 Signature: req.Signature,
240 attested, err := cfg.Runs.RecordAttestation(run.ID, owner, runs.Attestation{
241 Signer: rec.Signer,
242 Fingerprint: rec.Key,
243 CommitID: rec.Commit,
254 244 }, now)
255 245 if err != nil {
256 246 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
@@ -260,13 +250,13 @@ func (s *Server) serveRunner(ch ssh.Channel, user, accountID string) {
260 250 _ = enc.Encode(runnerReply{Op: "finish", OK: false})
261 251 continue
262 252 }
263 ok, err := cfg.Runs.Finish(req.Run, owner, req.Exit, now)
253 ok, err := cfg.Runs.Finish(run.ID, owner, rec.Exit, now)
264 254 if err != nil {
265 255 _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()})
266 256 continue
267 257 }
268 cfg.Log.Info("run finished", "user", user, "run", req.Run, "exit", req.Exit,
269 "signer", accountID, "key", fingerprint, "recorded", ok)
258 cfg.Log.Info("run finished", "user", user, "run", run.ID, "exit", rec.Exit,
259 "signer", rec.Signer, "key", rec.Key, "commit", rec.Commit, "recorded", ok)
270 260 _ = enc.Encode(runnerReply{Op: "finish", OK: ok})
271 261
272 262 case "ping":
@@ -285,122 +275,6 @@ func (s *Server) serveRunner(ch ssh.Channel, user, accountID string) {
285 275 }
286 276 }
287 277
288 // checkArtifacts validates what a runner claims it uploaded. The store is the
289 // authority on whether the bytes exist; this only refuses claims that are not
290 // downloadable at all.
291 // claimFor assembles the statement a runner signs: the job as the server recorded
292 // it, plus what the runner reports it produced. Both sides build it from one value,
293 // so a disagreement in either direction is a claim that does not verify.
294 func (s *Server) claimFor(req runnerRequest, arts []runs.Artifact) ([]byte, error) {
295 run, err := s.cfg.Runs.Get(req.Run)
296 if err != nil {
297 return nil, fmt.Errorf("reading run %s: %w", req.Run, err)
298 }
299 if run == nil {
300 return nil, fmt.Errorf("no run %s", req.Run)
301 }
302 return runs.Claim{
303 RunID: run.ID,
304 Repo: run.Repo,
305 Ref: run.Ref,
306 SHA: run.SHA,
307 Workflow: run.Workflow,
308 WorkflowSHA: run.WorkflowSHA,
309 Job: run.Job,
310 StepsSHA: run.StepsSHA,
311 Exit: req.Exit,
312 LogOID: req.LogOID,
313 LogSize: req.LogSize,
314 Artifacts: arts,
315 Finished: req.Finished,
316 }.Canonical()
317 }
318
319 // verifyClaim checks a signature over the claim against the keys of the account the
320 // channel authenticated as, and returns the fingerprint that matched.
321 //
322 // The account is not the claim's to choose: it is the identity the server resolved
323 // from the SSH key that opened the connection. The signing key may differ from the
324 // connecting one — an agent-held key to connect, a file to sign — but it must belong
325 // to the same account, or a runner could sign as somebody else.
326 func (s *Server) verifyClaim(accountID string, payload, sig []byte) (string, error) {
327 acct, err := s.cfg.Identity.Find(accountID)
328 if err != nil || acct == nil {
329 return "", fmt.Errorf("cannot read the account %s: %v", accountID, err)
330 }
331 for _, k := range acct.Keys {
332 pub, _, _, _, err := ssh.ParseAuthorizedKey([]byte(k.Line))
333 if err != nil {
334 continue
335 }
336 cpk, ok := pub.(ssh.CryptoPublicKey)
337 if !ok {
338 continue
339 }
340 ed, ok := cpk.CryptoPublicKey().(ed25519.PublicKey)
341 if !ok {
342 continue
343 }
344 if ed25519.Verify(ed, payload, sig) {
345 return k.Fingerprint, nil
346 }
347 }
348 return "", fmt.Errorf("the signature does not verify under any key of %s: sign with the key you push with", acct.Name)
349 }
350
351 // checkRunLog verifies a log claim the way checkArtifacts verifies an artifact
352 // claim: the bytes must be in the store, and the declared size must be theirs.
353 func (s *Server) checkRunLog(oid string, size int64) error {
354 if !lfs.ValidOID(oid) {
355 return fmt.Errorf("log: %q is not a sha256 object id", oid)
356 }
357 if s.cfg.LFS == nil {
358 return fmt.Errorf("this instance has no object store, so a log cannot be archived")
359 }
360 n, err := s.cfg.LFS.Stat(oid)
361 if err != nil {
362 return fmt.Errorf("log names object %s, which is not in the store: upload it before finishing", oid[:12])
363 }
364 if size != n {
365 return fmt.Errorf("log says %d bytes but the stored object is %d", size, n)
366 }
367 return nil
368 }
369
370 func (s *Server) checkArtifacts(in []runnerArtifact) ([]runs.Artifact, error) {
371 if len(in) > maxRunArtifacts {
372 return nil, fmt.Errorf("%d artifacts, limit %d", len(in), maxRunArtifacts)
373 }
374 seen := make(map[string]bool, len(in))
375 out := make([]runs.Artifact, 0, len(in))
376 for _, a := range in {
377 if err := runs.ValidArtifactName(a.Name); err != nil {
378 return nil, err
379 }
380 if seen[a.Name] {
381 return nil, fmt.Errorf("two artifacts are named %q", a.Name)
382 }
383 seen[a.Name] = true
384 if !lfs.ValidOID(a.OID) {
385 return nil, fmt.Errorf("artifact %q: %q is not a sha256 object id", a.Name, a.OID)
386 }
387 if s.cfg.LFS == nil {
388 return nil, fmt.Errorf("this instance has no object store, so artifacts cannot be published")
389 }
390 n, err := s.cfg.LFS.Stat(a.OID)
391 if err != nil {
392 return nil, fmt.Errorf("artifact %q names object %s, which is not in the store: "+
393 "upload it before finishing the run", a.Name, a.OID[:12])
394 }
395 if a.Size != n {
396 return nil, fmt.Errorf("artifact %q says %d bytes but the stored object is %d",
397 a.Name, a.Size, n)
398 }
399 out = append(out, runs.Artifact{Name: a.Name, OID: a.OID, Size: n})
400 }
401 return out, nil
402 }
403
404 278 // runnableRepos lists the repositories this account may execute jobs for.
405 279 func (s *Server) runnableRepos(accountID string) ([]string, error) {
406 280 repos, err := s.cfg.Repos.List()