1// SPDX-License-Identifier: MIT
2
3package guest
4
5import (
6 "context"
7 "fmt"
8 "net"
9 "os"
10 "os/user"
11 "strconv"
12 "strings"
13 "sync"
14 "syscall"
15 "time"
16
17 "github.com/bnema/zerowrap"
18
19 "bnema.dev/ermite-runner/internal/execstream"
20 "bnema.dev/ermite-runner/internal/execstream/execpb"
21 "bnema.dev/ermite-runner/internal/executor"
22)
23
24// frameReporter is the executor.Reporter that writes frames on the stream.
25type frameReporter struct {
26 conn *execstream.Conn
27 // outcomes keeps steps.<id>.outcome from StepResults until StepCompleted.
28 omu sync.Mutex
29 outcomes map[int]executor.Conclusion
30 // err keeps the first send error; the job goes on without a host.
31 failed chan struct{}
32}
33
34func ts(t time.Time) int64 { return t.UnixNano() }
35
36func conclusionProto(c executor.Conclusion) execpb.StepConclusion {
37 switch c {
38 case executor.ConclusionSuccess:
39 return execpb.StepConclusion_STEP_CONCLUSION_SUCCESS
40 case executor.ConclusionFailure:
41 return execpb.StepConclusion_STEP_CONCLUSION_FAILURE
42 case executor.ConclusionCancelled:
43 return execpb.StepConclusion_STEP_CONCLUSION_CANCELLED
44 case executor.ConclusionSkipped:
45 return execpb.StepConclusion_STEP_CONCLUSION_SKIPPED
46 case executor.ConclusionTimedOut:
47 return execpb.StepConclusion_STEP_CONCLUSION_TIMED_OUT
48 }
49 return execpb.StepConclusion_STEP_CONCLUSION_UNSPECIFIED
50}
51
52func (r *frameReporter) send(f *execpb.Frame) {
53 if err := r.conn.Send(f); err != nil {
54 select {
55 case <-r.failed:
56 default:
57 close(r.failed)
58 }
59 }
60}
61
62func (r *frameReporter) StepStarted(n int, name string, at time.Time) {
63 r.send(&execpb.Frame{Body: &execpb.Frame_StepStarted{StepStarted: &execpb.StepStarted{Number: int32(n), Name: name, UnixNano: ts(at)}}})
64}
65
66func (r *frameReporter) Log(n int, at time.Time, line string) {
67 for i, piece := range execstream.SplitLine(line, execstream.MaxLogLine) {
68 r.send(&execpb.Frame{Body: &execpb.Frame_Log{Log: &execpb.Log{Step: int32(n), UnixNano: ts(at), Text: piece, Continued: i > 0}}})
69 }
70}
71
72func (r *frameReporter) StepCompleted(n int, name string, c executor.Conclusion, at time.Time) {
73 r.omu.Lock()
74 outcome := r.outcomes[n]
75 delete(r.outcomes, n)
76 r.omu.Unlock()
77 r.send(&execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{
78 Number: int32(n), Name: name, Conclusion: conclusionProto(c), Outcome: conclusionProto(outcome), UnixNano: ts(at),
79 }}})
80}
81
82func levelProto(l executor.AnnotationLevel) execpb.AnnotationLevel {
83 switch l {
84 case executor.LevelNotice:
85 return execpb.AnnotationLevel_ANNOTATION_LEVEL_NOTICE
86 case executor.LevelWarning:
87 return execpb.AnnotationLevel_ANNOTATION_LEVEL_WARNING
88 case executor.LevelFailure:
89 return execpb.AnnotationLevel_ANNOTATION_LEVEL_FAILURE
90 }
91 return execpb.AnnotationLevel_ANNOTATION_LEVEL_UNSPECIFIED
92}
93
94// StepResults sends the annotations and the summary of a step, and keeps its
95// outcome for the StepCompleted frame that follows.
96func (r *frameReporter) StepResults(n int, res executor.StepResults) {
97 r.omu.Lock()
98 if r.outcomes == nil {
99 r.outcomes = map[int]executor.Conclusion{}
100 }
101 r.outcomes[n] = res.Outcome
102 r.omu.Unlock()
103 if len(res.Annotations) > 0 {
104 sa := &execpb.StepAnnotations{Step: int32(n)}
105 for _, a := range res.Annotations {
106 sa.Annotations = append(sa.Annotations, &execpb.Annotation{
107 Level: levelProto(a.Level), Title: a.Title, Message: a.Message, File: a.File,
108 StartLine: a.StartLine, EndLine: a.EndLine, StartColumn: a.StartColumn, EndColumn: a.EndColumn,
109 })
110 }
111 r.send(&execpb.Frame{Body: &execpb.Frame_StepAnnotations{StepAnnotations: sa}})
112 }
113 if res.HasSummary {
114 r.send(&execpb.Frame{Body: &execpb.Frame_StepSummary{StepSummary: &execpb.StepSummary{Step: int32(n), Markdown: res.Summary}}})
115 }
116}
117
118func (r *frameReporter) JobCompleted(res executor.Result) {
119 r.send(&execpb.Frame{Body: &execpb.Frame_JobResult{JobResult: &execpb.JobResult{
120 Conclusion: conclusionProto(res.Conclusion), Outputs: res.Outputs, Error: res.Error, RunnerShutdown: res.RunnerShutdown,
121 }}})
122}
123
124// StepUser is the account the steps run as: uid, gid and supplementary groups,
125// like the hosted runners' "runner" user. Nil credential means the steps run as
126// the executor itself.
127type StepUser struct {
128 Credential *syscall.Credential
129 Name, Home string
130}
131
132// LookupStepUser finds the user the steps run as. When the executor is not
133// root it cannot switch user: the steps run as the current user (a development
134// or test setup, never the sandbox of a runner, whose init is root). When it is
135// root and the user does not exist, the lookup fails: running a job's code as
136// root of the sandbox, with the docker daemon and the whole image at hand,
137// because an image lacks the account would hide a broken image. A test image
138// opts in with allowRoot.
139func LookupStepUser(name string, allowRoot bool) (StepUser, error) {
140 return lookupStepUser(os.Geteuid(), user.Lookup, name, allowRoot)
141}
142
143func lookupStepUser(euid int, lookup func(string) (*user.User, error), name string, allowRoot bool) (StepUser, error) {
144 if euid != 0 || name == "" {
145 u := StepUser{Home: os.Getenv("HOME")}
146 if cu, err := user.Current(); err == nil {
147 u.Name = cu.Username
148 if u.Home == "" {
149 u.Home = cu.HomeDir
150 }
151 }
152 return u, nil
153 }
154 u, err := lookup(name)
155 if err != nil {
156 if allowRoot {
157 return StepUser{Name: "root", Home: "/root"}, nil
158 }
159 return StepUser{}, fmt.Errorf("the image has no user %q to run the steps as (%v): refusing to run them as root", name, err)
160 }
161 uid, _ := strconv.ParseUint(u.Uid, 10, 32)
162 gid, _ := strconv.ParseUint(u.Gid, 10, 32)
163 cred := &syscall.Credential{Uid: uint32(uid), Gid: uint32(gid)}
164 if ids, err := u.GroupIds(); err == nil {
165 for _, g := range ids {
166 if n, err := strconv.ParseUint(g, 10, 32); err == nil {
167 cred.Groups = append(cred.Groups, uint32(n))
168 }
169 }
170 }
171 return StepUser{Credential: cred, Name: u.Username, Home: u.HomeDir}, nil
172}
173
174// stepEnv is the environment the steps inherit: the executor's, without the
175// variables of the stream, with HOME, USER and LOGNAME of the step user.
176func stepEnv(environ []string, u StepUser) []string {
177 var out []string
178 for _, kv := range environ {
179 k, _, _ := strings.Cut(kv, "=")
180 switch {
181 case strings.HasPrefix(k, "ERMITE_"), k == "RUNNER_TOOL_CACHE", k == "HOME", k == "USER", k == "LOGNAME":
182 continue
183 }
184 out = append(out, kv)
185 }
186 if u.Home != "" {
187 out = append(out, "HOME="+u.Home)
188 }
189 if u.Name != "" {
190 out = append(out, "USER="+u.Name, "LOGNAME="+u.Name)
191 }
192 return out
193}
194
195func cancelKind(k execpb.CancelKind) executor.CancelReason {
196 switch k {
197 case execpb.CancelKind_CANCEL_KIND_TIMEOUT:
198 return executor.CancelTimeout
199 case execpb.CancelKind_CANCEL_KIND_SHUTDOWN:
200 return executor.CancelShutdown
201 }
202 return executor.CancelUser
203}
204
205// runJob runs the job of a Run frame with the executor and reports on conn.
206// It returns the exit code to report.
207func runJob(ctx context.Context, conn *execstream.Conn, cfg ExecConfig, spec *execpb.JobSpec, vals jobValues) int {
208 su, err := LookupStepUser(cfg.StepUser, cfg.AllowRootSteps)
209 if err != nil {
210 // Reported like any failure of "Set up job": the job fails, with the reason
211 // in its log, and nothing of the workflow runs.
212 return failSetup(conn, err.Error())
213 }
214 canceller := executor.NewCanceller()
215 rep := &frameReporter{conn: conn, failed: make(chan struct{})}
216 res := newResolver(conn)
217 cacheEnv := startCacheRelay(ctx, cfg.Target, spec)
218 defer cacheEnv.close()
219
220 // Frames from the host while the job runs: a Cancel stops the job; a lost
221 // host does too, like a runner shutdown. The reader ends with the stream: the
222 // caller closes it after runJob, and the goroutine that reports a failed
223 // send ends with the job (done).
224 done := make(chan struct{})
225 defer close(done)
226 go func() {
227 for {
228 f, err := conn.Recv()
229 if err != nil {
230 canceller.Cancel(executor.CancelShutdown)
231 return
232 }
233 if c := f.GetCancel(); c != nil {
234 canceller.Cancel(cancelKind(c.GetKind()))
235 }
236 if r := f.GetResolveResponse(); r != nil {
237 res.deliver(r)
238 }
239 }
240 }()
241 // The executor's own context (SIGTERM forwarded by init: the sandbox is
242 // being stopped) is a shutdown.
243 stop := context.AfterFunc(ctx, func() { canceller.Cancel(executor.CancelShutdown) })
244 defer stop()
245 go func() {
246 select {
247 case <-rep.failed:
248 canceller.Cancel(executor.CancelShutdown)
249 case <-canceller.Done():
250 case <-done:
251 }
252 }()
253
254 env := spec.GetRunnerEnvironment()
255 result := executor.Run(canceller, executor.Config{
256 WorkflowPath: spec.GetWorkflowPath(), WorkflowYAML: []byte(spec.GetWorkflowYaml()), JobKey: spec.GetJobKey(),
257 ContextsJSON: spec.GetContexts(), MatrixJSON: spec.GetMatrix(), Repo: spec.GetRepo(),
258 TimeoutMinutes: int(spec.GetTimeoutMinutes()), Deadline: time.Duration(spec.GetDeadlineMs()) * time.Millisecond,
259 Runner: executor.RunnerInfo{Name: spec.GetRunnerName(), Environment: env},
260 RunnerVersion: spec.GetRunnerVersion(),
261 Home: su.Home, ToolCache: os.Getenv("RUNNER_TOOL_CACHE"), BaseEnv: stepEnv(os.Environ(), su), Credential: su.Credential,
262 CacheURL: cacheEnv.url, CacheToken: cacheEnv.token,
263 Debug: spec.GetDebug(), Reporter: rep, Actions: executor.DefaultActionsConfig(res.resolve), Log: zerowrap.FromCtx(ctx),
264 Secrets: vals.secrets, Vars: vals.vars, SecretsWithheld: spec.GetSecretsWithheld(), TokenPermissions: tokenPermissions(spec),
265 SendAnnotations: spec.GetSendAnnotations(), SendSummaries: spec.GetSendSummaries(),
266 TermAfter: cfg.TermAfter, KillAfter: cfg.KillAfter, WaitDelay: cfg.WaitDelay,
267 })
268 if result.Conclusion == executor.ConclusionSuccess {
269 return 0
270 }
271 return 1
272}
273
274// cacheURL is ACTIONS_CACHE_URL: the cache service of the runner on the
275// loopback of the sandbox, where the relay listens. The trailing slash matters:
276// the toolkit appends "_apis/…" to it (cacheHttpClient.ts:47). Empty when the
277// job has no cache service.
278func cacheURL(port uint32) string {
279 if port == 0 || port > 65535 {
280 return ""
281 }
282 return "http://" + net.JoinHostPort(CacheListenHost, strconv.FormatUint(uint64(port), 10)) + "/"
283}
284
285// cacheRelay is the cache of a job in the sandbox: the relay, the URL of the
286// loopback it listens on and the token of the job. All are empty without a cache
287// service.
288type cacheRelay struct {
289 relay *Relay
290 url, token string
291}
292
293func (c cacheRelay) close() {
294 if c.relay != nil {
295 _ = c.relay.Close()
296 }
297}
298
299// startCacheRelay starts the relay of the cache for a job that has a cache
300// service: it listens on the loopback port the host says and pipes every
301// connection to a new stream on the cache channel of the host (a sibling of
302// the executor's target). A relay that cannot start leaves the job without a
303// cache, like a scope the host refuses: the job runs.
304func startCacheRelay(ctx context.Context, target Target, spec *execpb.JobSpec) cacheRelay {
305 url := cacheURL(spec.GetCachePort())
306 if url == "" || spec.GetCacheToken() == "" {
307 return cacheRelay{}
308 }
309 addr := net.JoinHostPort(CacheListenHost, strconv.FormatUint(uint64(spec.GetCachePort()), 10))
310 r, err := StartRelay(ctx, addr, target.CacheTarget().DialStream, DefaultRelayMax)
311 if err != nil {
312 log := zerowrap.FromCtx(ctx)
313 log.Warn().Err(err).Msg("job runs without the cache")
314 return cacheRelay{}
315 }
316 return cacheRelay{relay: r, url: url, token: spec.GetCacheToken()}
317}
318
319// failSetup ends a job that cannot start, like a failed "Set up job": step 1
320// starts, logs the reason as an error, and fails; the job result is a failure.
321func failSetup(conn *execstream.Conn, msg string) int {
322 rep := &frameReporter{conn: conn, failed: make(chan struct{})}
323 now := time.Now()
324 rep.StepStarted(1, executor.SetUpJobName, now)
325 rep.Log(1, now, "##[error]"+msg)
326 rep.StepCompleted(1, executor.SetUpJobName, executor.ConclusionFailure, time.Now())
327 rep.JobCompleted(executor.Result{Conclusion: executor.ConclusionFailure, Error: msg})
328 return 1
329}
330
331// tokenPermissions is the permissions of GITHUB_TOKEN the server sent, nil when
332// it sent none (the executor then works them out from the workflow).
333func tokenPermissions(spec *execpb.JobSpec) map[string]string {
334 p := spec.GetTokenPermissions()
335 if p == nil {
336 return nil
337 }
338 if p.GetLevels() == nil {
339 return map[string]string{}
340 }
341 return p.GetLevels()
342}