1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "crypto/rand"
8 "encoding/base64"
9 "encoding/hex"
10 "fmt"
11 "slices"
12 "strings"
13 "sync"
14 "time"
15
16 "github.com/bnema/zerowrap"
17
18 "bnema.dev/ermite/internal/domain"
19 "bnema.dev/ermite/internal/ports"
20)
21
22// Defaults of the CI engine.
23const (
24 // DefaultCIPollTimeout is how long FetchJob holds a call: the contract's 50 s.
25 DefaultCIPollTimeout = 50 * time.Second
26 // DefaultCITick is how often the watchdog looks at the jobs in flight.
27 DefaultCITick = time.Second
28 // DefaultCIKeepalive is how often WatchJob sends a keepalive.
29 DefaultCIKeepalive = 30 * time.Second
30 // DefaultCITriggerQueue is the size of the queue of pushes to plan.
31 DefaultCITriggerQueue = 256
32 // DefaultCIRemoteConcurrency is the git ls-remote processes ResolveActions
33 // runs at once, whatever the number of jobs that resolve.
34 DefaultCIRemoteConcurrency = 4
35 // DefaultCIResolveTimeout bounds one ResolveActions call.
36 DefaultCIResolveTimeout = 60 * time.Second
37 // DefaultCIConcurrencyPruneBatch is the stale marks of a concurrency group a
38 // transaction removes at most.
39 DefaultCIConcurrencyPruneBatch = 256
40 // MaxCIChecks is the most checks ListChecks returns.
41 MaxCIChecks = 1000
42 // MaxCILogStreams and MaxCILogStreamsPerClient bound the concurrent log
43 // streams (GetJobLog).
44 MaxCILogStreams = 64
45 MaxCILogStreamsPerClient = 8
46 // maxCIWorkflowBytes is the largest workflow file (the parser's limit).
47 maxCIWorkflowBytes = 1 << 20
48 // maxCIWorkflowFiles bounds the workflow files read per push.
49 maxCIWorkflowFiles = 256
50)
51
52// CIConfig holds what the CI use cases need. Store, Logs, Summaries, Meta, Repos, Reader,
53// Git, Diff, Bus, Clock, Owner and PublicURL are required (the config supplies
54// the defaults of Owner and Retention).
55type CIConfig struct {
56 Store ports.CIStore
57 Logs ports.CILogStore
58 // Summaries keeps the job summaries, deleted with the logs of their run.
59 Summaries ports.CISummaryStore
60 Meta ports.RepoMetaStore
61 Repos ports.GitRepos
62 Reader ports.RepoReader
63 Git ports.MergeGit
64 Diff ports.CIGit
65 // MRs, Pull and Committer give the workflows their pull_request event and
66 // the refs/pull/<n>/{head,merge} refs of the merge requests (CI-3, D5);
67 // without MRs or Pull the merge requests start no run and have no such
68 // refs. Committer is the identity of the test merge commits (default
69 // "ermite"). Fresh, when set, is told when the refs of a repository changed
70 // so that the reader sees them at once.
71 MRs ports.MergeRequestStore
72 Pull ports.PullGit
73 Committer domain.Identity
74 Fresh ports.RepoFreshener
75 // PullStage, when set, is called by the workers of the merge requests at
76 // the stages "event" (an activity is taken), "computed" (its merge is
77 // computed, before it is published) and "refresh" (a refresh is taken).
78 // Tests use it to interleave other changes.
79 PullStage func(stage string)
80 // PullRefreshes bounds the merge requests waiting for a refresh of their refs
81 // (default DefaultCIPullRefreshes); past it repositories are rescanned. It
82 // does not bound the activities (synchronize runs), which have their own queue
83 // with the overload policy of the push triggers.
84 PullRefreshes int
85 // PullAttempts bounds the merge requests whose last conflict is remembered
86 // (default DefaultCIPullAttempts), so that a rescan does not try a conflicting
87 // merge again until a branch moves.
88 PullAttempts int
89 // PullSnapshot bounds the merge requests one push takes the snapshot of, in the
90 // push handler (default DefaultCIPullSnapshot): past it the others are skipped
91 // with an error in the log and the repository is rescanned for its refs.
92 PullSnapshot int
93 Bus ports.EventBus
94 Clock ports.Clock
95 // Auth re-validates the token of a log stream; nil skips it.
96 Auth ports.Revalidator
97 // Remote lists the refs of github.com repositories for ResolveActions; nil
98 // answers every github.com action with unavailable.
99 Remote ports.GitRemote
100 // Workflows downloads reusable workflow files from github.com at a resolved
101 // commit; nil fails such a call as unavailable.
102 Workflows ports.WorkflowFetcher
103 // Archive serves the tarballs of the REST subset.
104 Archive ports.GitArchive
105 // Tokens finds the stored token of the pusher of a run, to ask what it may
106 // read when an action of another repository is resolved for the run. Nil
107 // reads nothing for a token (fail closed).
108 Tokens ports.TokenStore
109 // Settings resolves the variables and secrets of a job (CI-3). Nil gives
110 // jobs none, and an empty "vars" context. Leave it unset rather than
111 // assigning a typed nil pointer.
112 Settings ports.CIJobSettings
113 // RemoteConcurrency bounds the git ls-remote processes at once (default
114 // DefaultCIRemoteConcurrency); ResolveTimeout bounds one ResolveActions call
115 // (default DefaultCIResolveTimeout).
116 RemoteConcurrency int
117 ResolveTimeout time.Duration
118 // Owner is the owner login of the github context.
119 Owner string
120 // PublicURL is the base URL of the instance (github.server_url).
121 PublicURL string
122 // Retention is how long runs, jobs, checks and logs are kept.
123 Retention time.Duration
124
125 // The durations below default to the contract's; tests shorten them.
126 PollTimeout time.Duration
127 Tick time.Duration
128 Keepalive time.Duration
129 AckTimeout time.Duration
130 HeartbeatDeadline time.Duration
131 CancelGrace time.Duration
132 TimeoutSlack time.Duration
133 QueueTimeout time.Duration
134 TriggerQueue int
135 // ConcurrencyRetries bounds the retries of a delivery of the concurrency
136 // groups that failed (a cancel, the planning of a promoted entry): 0 is the
137 // default (8), a negative number none, which the tests use to leave the
138 // recovery to the reconciliation.
139 ConcurrencyRetries int
140 // ConcurrencyPruneBatch bounds the stale marks of a concurrency group one
141 // transaction removes (default 256, far below the size limit of a kv batch);
142 // the others go with the next decisions and the reconciliation.
143 ConcurrencyPruneBatch int
144 // ScheduleTick is how often the scheduler looks at the clock (default one
145 // second) and ScheduleBudget the repositories it handles per tick (default
146 // 64); see ci_schedule.go.
147 ScheduleTick time.Duration
148 ScheduleBudget int
149 // RefTips holds who moves each branch and tag to its object, written before
150 // the ref moves (Pushes, MergeRequests), for the actor and trust of scheduled
151 // runs and the trust of dispatched ones (ci_tips.go). Nil: they are never
152 // trusted. Leave it unset rather than assigning a typed nil pointer.
153 RefTips ports.CIRefTipStore
154 // DispatchResolved, when set, is called by DispatchWorkflow after it resolved
155 // its ref and before it judges the trust of the run. Tests use it to move the
156 // ref in between.
157 DispatchResolved func()
158 // RunEventHandled, when set, is called after the workflows waiting for a
159 // "requested" or "completed" event of a run were started (or none matched).
160 // Tests use it as a barrier.
161 RunEventHandled func(runID int64, action string)
162 // ScheduleVisit, when set, is called with each repository the scheduler
163 // looks at. Tests use it.
164 ScheduleVisit func(repo domain.RepoName)
165}
166
167func (c CIConfig) withDefaults() CIConfig {
168 def := func(d *time.Duration, v time.Duration) {
169 if *d <= 0 {
170 *d = v
171 }
172 }
173 def(&c.PollTimeout, DefaultCIPollTimeout)
174 def(&c.Tick, DefaultCITick)
175 def(&c.Keepalive, DefaultCIKeepalive)
176 def(&c.AckTimeout, domain.CIAckTimeout)
177 def(&c.HeartbeatDeadline, domain.CIHeartbeatDeadline)
178 def(&c.CancelGrace, domain.CICancelGrace)
179 def(&c.TimeoutSlack, domain.CITimeoutSlack)
180 def(&c.QueueTimeout, domain.CIQueueTimeout)
181 def(&c.Retention, domain.CIRetentionDays*24*time.Hour)
182 if c.RemoteConcurrency <= 0 {
183 c.RemoteConcurrency = DefaultCIRemoteConcurrency
184 }
185 def(&c.ResolveTimeout, DefaultCIResolveTimeout)
186 if c.PullSnapshot <= 0 {
187 c.PullSnapshot = DefaultCIPullSnapshot
188 }
189 if c.PullRefreshes <= 0 {
190 c.PullRefreshes = DefaultCIPullRefreshes
191 }
192 if c.TriggerQueue <= 0 {
193 c.TriggerQueue = DefaultCITriggerQueue
194 }
195 if c.ConcurrencyPruneBatch <= 0 {
196 c.ConcurrencyPruneBatch = DefaultCIConcurrencyPruneBatch
197 }
198 def(&c.ScheduleTick, DefaultCIScheduleTick)
199 if c.ScheduleBudget <= 0 {
200 c.ScheduleBudget = DefaultCIScheduleBudget
201 }
202 return c
203}
204
205// CI is the CI use cases: the workflow trigger, the planner, the assignment of
206// jobs to runners, the runner protocol and the owner side (ADR 0025).
207//
208// Every change of a run goes through its lock (withRun): the records of a run
209// are read, changed and written again by one goroutine at a time, so that two
210// calls (a runner's update and an owner's cancel) never lose one another's
211// write. Assignment takes the assignment lock before the lock of the run.
212type CI struct {
213 cfg CIConfig
214 boot time.Time
215 queue chan ciTrigger
216 // runEvents are the requested and completed events of runs, for the
217 // workflow_run workflows (ci_workflowrun.go).
218 runEvents chan runEvent
219 sched *scheduler // the schedule trigger (ci_schedule.go)
220 pull *pullQueue // the work of the merge requests (ci_pull.go)
221
222 pullFileLists pullFilesCache // the files of the merge requests (ci_pullfiles.go)
223
224 stop context.CancelFunc
225 ctx context.Context
226 wg sync.WaitGroup
227
228 remoteSlots chan struct{} // slots of the git ls-remote processes
229 locks sync.Map // run id -> *sync.Mutex
230 assign sync.Mutex
231 jobs notifier // woken when a job changes: key is the job reference
232 pending notifier // woken when a job may be ready or a runner was removed
233 flows *wfCache
234
235 mu sync.Mutex
236 heartbeats map[string]time.Time // job reference -> last heartbeat
237 seen map[string]time.Time // runner id -> last seen
238 polls map[string]pollRecord
239 // archiveTokens are the secrets of the tokens given for the archives of
240 // ermite actions, per job reference then "<repository>@<sha>", so that a job
241 // that resolves the same commit twice gets one token (memory only: a
242 // restart issues new ones).
243 archiveTokens map[string]map[string]string
244 streams int
245 perClient map[string]int
246 closed bool
247}
248
249var (
250 _ ports.RunnerService = (*CI)(nil)
251 _ ports.RunService = (*CI)(nil)
252 _ ports.WorkflowTrigger = (*CI)(nil)
253 _ ports.CIChecksReader = (*CI)(nil)
254)
255
256// pollRecord remembers the assignment a poll made, for the retry of a lost
257// response (FetchJobRequest.poll_id). The token is kept in memory only.
258type pollRecord struct {
259 pollID string
260 ref string
261 token string
262 ghToken string
263 hash domain.TokenHash
264}
265
266// NewCI builds the CI use cases. ctx carries the logger; its cancellation is
267// not followed: call Start to run the workers and Close to stop them.
268func NewCI(ctx context.Context, cfg CIConfig) *CI {
269 if cfg.Owner == "" {
270 panic("usecase: CIConfig.Owner is required")
271 }
272 cfg = cfg.withDefaults()
273 run, stop := context.WithCancel(context.WithoutCancel(ctx))
274 return &CI{
275 cfg: cfg, boot: cfg.Clock.Now(), queue: make(chan ciTrigger, cfg.TriggerQueue), runEvents: make(chan runEvent, cfg.TriggerQueue), sched: newScheduler(cfg.Clock.Now()), pull: newPullQueue(cfg.TriggerQueue, cfg.PullRefreshes, cfg.PullAttempts),
276 ctx: run, stop: stop, flows: newWfCache(64), remoteSlots: make(chan struct{}, cfg.RemoteConcurrency),
277 heartbeats: map[string]time.Time{}, seen: map[string]time.Time{}, polls: map[string]pollRecord{},
278 archiveTokens: map[string]map[string]string{},
279 perClient: map[string]int{},
280 }
281}
282
283// Start runs the trigger worker and the watchdog, and plans again the runs
284// that were open when the server stopped.
285func (c *CI) Start() {
286 c.wg.Add(6)
287 go c.scheduleLoop()
288 go c.runEventLoop()
289 go c.pullLoop()
290 go c.refreshLoop()
291 go c.triggerLoop()
292 go c.watchdog()
293 c.wg.Add(1)
294 go func() {
295 defer c.wg.Done()
296 c.resume()
297 }()
298}
299
300// Close stops the workers and waits for them.
301func (c *CI) Close(_ context.Context) {
302 c.mu.Lock()
303 c.closed = true
304 c.mu.Unlock()
305 c.stop()
306 c.wg.Wait()
307}
308
309func (c *CI) now() time.Time { return c.cfg.Clock.Now().UTC() }
310
311// --- identifiers and secrets ---
312
313func newToken(prefix string) string {
314 var raw [32]byte
315 _, _ = rand.Read(raw[:])
316 return prefix + base64.RawURLEncoding.EncodeToString(raw[:])
317}
318
319func newRunnerID() string {
320 var raw [8]byte
321 _, _ = rand.Read(raw[:])
322 return hex.EncodeToString(raw[:])
323}
324
325func jobRef(j domain.CIJob) string { return fmt.Sprintf("%d/%d/%s", j.RunID, j.RunAttempt, j.ID) }
326
327// --- locks and notifications ---
328
329// notifier wakes the goroutines waiting on a key. A channel exists only while
330// somebody waits on it.
331type notifier struct {
332 mu sync.Mutex
333 ch map[string]chan struct{}
334}
335
336// wait returns a channel closed by the next notify of key.
337func (n *notifier) wait(key string) <-chan struct{} {
338 n.mu.Lock()
339 defer n.mu.Unlock()
340 if n.ch == nil {
341 n.ch = map[string]chan struct{}{}
342 }
343 c, ok := n.ch[key]
344 if !ok {
345 c = make(chan struct{})
346 n.ch[key] = c
347 }
348 return c
349}
350
351func (n *notifier) notify(key string) {
352 n.mu.Lock()
353 defer n.mu.Unlock()
354 if c, ok := n.ch[key]; ok {
355 close(c)
356 delete(n.ch, key)
357 }
358}
359
360const queueKey = "q"
361
362func (c *CI) wakeQueue() { c.pending.notify(queueKey) }
363
364func (c *CI) runLock(runID int64) *sync.Mutex {
365 m, _ := c.locks.LoadOrStore(runID, &sync.Mutex{})
366 return m.(*sync.Mutex)
367}
368
369// --- run state ---
370
371// runState is a run being changed under its lock: the run, the jobs of one of
372// its attempts, and what must be written. flush writes everything in one batch
373// and publishes the events.
374type runState struct {
375 c *CI
376 run domain.CIRun
377 attempt int32
378 jobs []domain.CIJob
379 src *domain.CIRunSource
380
381 runDirty bool
382 dirty map[string]bool
383 // chk marks jobs whose check must be written even when the job itself did
384 // not change (a rerun moves every check to the new attempt).
385 checks map[string]bool
386 // tokens are the job repository tokens to create, revoke the jobs whose
387 // tokens are deleted, with the next flush.
388 tokens []domain.CIRepoToken
389 revoke []ports.CIJobRef
390 // resolutions are the action resolutions to keep with the next flush.
391 resolutions []domain.CIActionResolution
392 // annotations replace the annotations of steps, clear drops all those of
393 // executions of jobs, with the next flush.
394 annotations []ports.CIStepAnnotations
395 clear []ports.CILogRef
396}
397
398// withRun locks the run, loads it with the jobs of attempt (0 for the latest)
399// and calls fn. The run is found by id alone through its source.
400func (c *CI) withRun(ctx context.Context, runID int64, attempt int32, fn func(rs *runState) error) error {
401 lock := c.runLock(runID)
402 lock.Lock()
403 defer lock.Unlock()
404 rs, err := c.loadRun(ctx, runID, attempt)
405 if err != nil {
406 return err
407 }
408 return fn(rs)
409}
410
411func (c *CI) loadRun(ctx context.Context, runID int64, attempt int32) (*runState, error) {
412 src, err := c.cfg.Store.GetRunSource(ctx, runID)
413 if err != nil {
414 return nil, err
415 }
416 run, err := c.cfg.Store.GetRun(ctx, src.Repo, runID)
417 if err != nil {
418 return nil, err
419 }
420 if attempt <= 0 {
421 attempt = run.Attempt
422 }
423 jobs, err := c.cfg.Store.ListJobs(ctx, runID, attempt)
424 if err != nil {
425 return nil, err
426 }
427 return &runState{c: c, run: run, attempt: attempt, jobs: jobs, src: &src, dirty: map[string]bool{}, checks: map[string]bool{}}, nil
428}
429
430// latest reports whether the state holds the latest attempt of the run.
431func (rs *runState) latest() bool { return rs.attempt == rs.run.Attempt }
432
433func (rs *runState) find(id string) *domain.CIJob {
434 for i := range rs.jobs {
435 if rs.jobs[i].ID == id {
436 return &rs.jobs[i]
437 }
438 }
439 return nil
440}
441
442func (rs *runState) touch(j *domain.CIJob) { rs.dirty[j.ID] = true }
443
444// checkOf builds the check of a job.
445func (rs *runState) checkOf(j domain.CIJob) domain.CICheck {
446 return domain.CICheck{
447 ID: j.ID, Repo: rs.run.Repo, Sha: rs.run.CheckSha(), Name: j.Name, Status: j.Status(), Conclusion: j.Conclusion,
448 RunID: rs.run.ID, JobID: j.ID, Attempt: rs.run.Attempt, WorkflowName: rs.run.WorkflowName,
449 CreatedAt: j.CreatedAt, StartedAt: j.StartedAt, CompletedAt: j.CompletedAt,
450 }
451}
452
453// flush recomputes the status of the run, writes the changed jobs, their
454// checks and the run in one batch, and publishes RunUpdated and JobUpdated.
455func (rs *runState) flush(ctx context.Context) error {
456 c := rs.c
457 now := c.now()
458 var w ports.CIWrite
459 var changedJobs []int
460 completing := false
461 for i := range rs.jobs {
462 if rs.dirty[rs.jobs[i].ID] {
463 w.Jobs = append(w.Jobs, rs.jobs[i])
464 changedJobs = append(changedJobs, i)
465 }
466 }
467 if rs.latest() {
468 for i := range rs.jobs {
469 if rs.dirty[rs.jobs[i].ID] || rs.checks[rs.jobs[i].ID] {
470 w.Checks = append(w.Checks, rs.checkOf(rs.jobs[i]))
471 }
472 }
473 if next := rs.recompute(now); next != rs.run {
474 completing = rs.run.Status != domain.CICompleted && next.Status == domain.CICompleted
475 rs.run, rs.runDirty = next, true
476 }
477 }
478 if rs.runDirty {
479 w.Runs = []domain.CIRun{rs.run}
480 }
481 w.RepoTokens, w.RevokeJobs, w.Resolutions = rs.tokens, rs.revoke, rs.resolutions
482 w.Annotations, w.ClearAnnotations = rs.annotations, rs.clear
483 if len(w.Jobs) == 0 && len(w.Runs) == 0 && len(w.Checks) == 0 && len(w.RepoTokens) == 0 && len(w.RevokeJobs) == 0 &&
484 len(w.Resolutions) == 0 && len(w.Annotations) == 0 && len(w.ClearAnnotations) == 0 {
485 return nil
486 }
487 v, err := c.cfg.Store.Apply(ctx, w)
488 if err != nil {
489 return fmt.Errorf("write run %d: %w", rs.run.ID, err)
490 }
491 for _, i := range changedJobs {
492 rs.jobs[i].Version = v
493 }
494 if rs.runDirty {
495 rs.run.Version = v
496 }
497 changedRun := rs.runDirty
498 rs.releaseEnded(ctx, changedJobs, changedRun && rs.run.Status == domain.CICompleted)
499 rs.dirty, rs.checks, rs.runDirty = map[string]bool{}, map[string]bool{}, false
500 rs.tokens, rs.revoke, rs.resolutions = nil, nil, nil
501 rs.annotations, rs.clear = nil, nil
502
503 repo, err := c.cfg.Meta.Get(ctx, rs.run.Repo)
504 if err != nil {
505 for _, i := range changedJobs {
506 c.jobs.notify(jobRef(rs.jobs[i]))
507 }
508 return nil // the repository is gone: nobody to tell
509 }
510 for _, i := range changedJobs {
511 j := rs.jobs[i]
512 lines, _ := c.cfg.Logs.Count(ctx, ports.CILogRef{RunID: j.RunID, JobID: j.ID, Attempt: j.Attempt})
513 c.cfg.Bus.Publish(ctx, domain.JobUpdated{Job: j, Lines: lines, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
514 c.jobs.notify(jobRef(j))
515 if j.Phase == domain.CIPhaseReady {
516 c.wakeQueue()
517 }
518 }
519 if changedRun {
520 c.cfg.Bus.Publish(ctx, domain.RunUpdated{Run: rs.run, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
521 }
522 if completing {
523 c.onRunEvent(ctx, rs.run, runCompleted)
524 }
525 return nil
526}
527
528// recompute returns the run with the status, the conclusion and the
529// timestamps its jobs imply.
530func (rs *runState) recompute(now time.Time) domain.CIRun {
531 run := rs.run
532 if run.Status == domain.CICompleted && len(rs.jobs) == 0 {
533 return run // an invalid workflow file: failed from the start
534 }
535 allEnded := len(rs.jobs) > 0
536 started := time.Time{}
537 var failed, cancelled bool
538 for _, j := range rs.jobs {
539 if !j.Ended() {
540 allEnded = false
541 }
542 // A job kept from an earlier attempt did not start in this one.
543 if j.Attempt == rs.attempt && !j.StartedAt.IsZero() && (started.IsZero() || j.StartedAt.Before(started)) {
544 started = j.StartedAt
545 }
546 switch {
547 case j.Conclusion.IsFailure() && j.ContinueOnError:
548 // continue-on-error: the job failed, the run does not.
549 case j.Conclusion.IsFailure():
550 failed = true
551 case j.Conclusion == domain.CICancelled:
552 cancelled = true
553 }
554 }
555 if run.StartedAt.IsZero() && !started.IsZero() {
556 run.StartedAt = started
557 }
558 switch {
559 case allEnded:
560 run.Status, run.CompletedAt = domain.CICompleted, now
561 switch {
562 case run.CancelRequested:
563 run.Conclusion = domain.CICancelled
564 case failed:
565 run.Conclusion = domain.CIFailure
566 case cancelled:
567 run.Conclusion = domain.CICancelled
568 default:
569 run.Conclusion = domain.CISuccess
570 }
571 if rs.run.Status == domain.CICompleted {
572 run.CompletedAt = rs.run.CompletedAt
573 }
574 case !run.StartedAt.IsZero():
575 run.Status, run.Conclusion, run.CompletedAt = domain.CIInProgress, domain.CINoConclusion, time.Time{}
576 default:
577 run.Status, run.Conclusion, run.CompletedAt = domain.CIQueued, domain.CINoConclusion, time.Time{}
578 }
579 return run
580}
581
582// --- access ---
583
584// readableRepo looks the repository up and applies the read rule.
585func (c *CI) readableRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
586 repo, err := lookupRepo(ctx, c.cfg.Meta, name)
587 if err != nil {
588 return domain.Repo{}, err
589 }
590 if !canRead(p, repo) {
591 return domain.Repo{}, domain.ErrNotFound
592 }
593 return repo, nil
594}
595
596// writableRepo is the access rule of cancelling and re-running: a principal
597// whose token holds ci:write on the repository sees the repository for this
598// action even without repo:read, as for merge requests.
599func (c *CI) writableRepo(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
600 repo, err := lookupRepo(ctx, c.cfg.Meta, name)
601 if err != nil {
602 return domain.Repo{}, err
603 }
604 if !canRead(p, repo) && !p.Allows(domain.ScopeCIWrite, name) {
605 return domain.Repo{}, domain.ErrNotFound
606 }
607 if err := scopeAccess(p, repo, domain.ScopeCIWrite); err != nil {
608 return domain.Repo{}, err
609 }
610 return repo, nil
611}
612
613// runWriteAccess is the branch rule of the write actions on a run, cancel and
614// re-run, which come after the ci:write check. A run is tied to the ref that
615// made it, and re-running it executes the workflow of that ref again while
616// cancelling it stops the checks of that ref. So, like a push, a merge or a
617// branch delete, these actions fall under the branch prefixes of a limited
618// token (ADR 0027): it may act only on the runs of branches under its prefixes,
619// and never on a run of a tag or of any other ref. Without prefixes, every run.
620//
621// Reading runs, jobs, logs and checks is not a write: it needs only the access
622// to read the repository, as for the code, whatever the prefixes. The caller
623// has seen the run by then, so the refusal is forbidden, not not found.
624func runWriteAccess(p domain.Principal, run domain.CIRun) error {
625 if !p.AllowsRef(run.Ref) {
626 return fmt.Errorf("%w: this token may not write %s", domain.ErrForbidden, run.Ref)
627 }
628 return nil
629}
630
631// canManageRunners is the access rule of runners: runner:register on a token
632// that is not restricted to repositories (a runner serves every repository),
633// or root. A token limited to branch prefixes never manages runners, whatever
634// scopes it holds: a runner runs the workflows of every branch of every
635// repository, with their code and secrets, so registering or removing one
636// reaches far beyond the prefixes (a replaced runner could also be a way to
637// take over the jobs of main). Like repository administration and owner
638// rights, it is out of reach of a limited token.
639func canManageRunners(p domain.Principal) error {
640 switch {
641 case p.IsAnonymous():
642 return domain.ErrUnauthenticated
643 case !p.HasScope(domain.ScopeRunnerRegister):
644 return fmt.Errorf("%w: this needs the %s scope", domain.ErrForbidden, domain.ScopeRunnerRegister)
645 case len(p.Repos()) > 0:
646 return fmt.Errorf("%w: a runner serves every repository: the token must not be restricted to some", domain.ErrForbidden)
647 case p.IsBranchLimited():
648 return fmt.Errorf("%w: a token limited to branch prefixes may not manage runners", domain.ErrForbidden)
649 }
650 return nil
651}
652
653// --- small helpers ---
654
655func truncateRunes(s string, n int) string {
656 if len(s) <= n {
657 return s
658 }
659 r := []rune(s)
660 if len(r) <= n {
661 return s
662 }
663 return string(r[:n])
664}
665
666// cutText cuts s to n bytes at a character boundary.
667func cutText(s string, n int) string {
668 if len(s) <= n {
669 return s
670 }
671 for n > 0 && n < len(s) && s[n]&0xC0 == 0x80 {
672 n--
673 }
674 return s[:n]
675}
676
677func sortedKeys[V any](m map[string]V) []string {
678 keys := make([]string, 0, len(m))
679 for k := range m {
680 keys = append(keys, k)
681 }
682 slices.Sort(keys)
683 return keys
684}
685
686// ctxLog returns the logger of the context, addressable so that the level
687// methods can be chained on it.
688func ctxLog(ctx context.Context) *zerowrap.Logger {
689 l := zerowrap.FromCtx(ctx)
690 return &l
691}
692
693// logRun returns the logger of the context with the fields of the run.
694func logRun(ctx context.Context, run domain.CIRun) zerowrap.Logger {
695 return zerowrap.FromCtx(ctx).WithFields(map[string]any{"repo": run.Repo.String(), "run_id": run.ID})
696}
697
698func isHex40(s string) bool {
699 if len(s) != 40 {
700 return false
701 }
702 return strings.Trim(s, "0123456789abcdef") == ""
703}