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, docker0IPv4)
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, CacheMode: cacheEnv.mode,
263 ContainerCacheURL: cacheEnv.containerURL, Results: cacheEnv.results, CacheOff: cacheEnv.cacheOff,
264 Debug: spec.GetDebug(), Reporter: rep, Actions: executor.DefaultActionsConfig(res.resolve), Log: zerowrap.FromCtx(ctx),
265 Secrets: vals.secrets, Vars: vals.vars, SecretsWithheld: spec.GetSecretsWithheld(), TokenPermissions: tokenPermissions(spec),
266 SendAnnotations: spec.GetSendAnnotations(), SendSummaries: spec.GetSendSummaries(),
267 TermAfter: cfg.TermAfter, KillAfter: cfg.KillAfter, WaitDelay: cfg.WaitDelay,
268 IDTokenURL: spec.GetIdTokenUrl(), IDTokenRequestToken: spec.GetIdTokenRequestToken(),
269 })
270 if result.Conclusion == executor.ConclusionSuccess {
271 return 0
272 }
273 return 1
274}
275
276// cacheURL is ACTIONS_CACHE_URL: the cache service of the runner on the
277// loopback of the sandbox, where the relay listens. The trailing slash matters:
278// the toolkit appends "_apis/…" to it (cacheHttpClient.ts:47). Empty when the
279// job has no cache service.
280func cacheURL(port uint32) string {
281 return relayURL(CacheListenHost, port)
282}
283
284// relayURL is the URL of the relay on host and port, "" for a port that is
285// not one.
286func relayURL(host string, port uint32) string {
287 if port == 0 || port > 65535 {
288 return ""
289 }
290 return "http://" + net.JoinHostPort(host, strconv.FormatUint(uint64(port), 10)) + "/"
291}
292
293// BridgeInterface is the default bridge of the sandbox's docker daemon. Its
294// address is a local address of the sandbox that every container of that
295// daemon reaches, whatever its bridge network: the traffic to it goes through
296// INPUT, which Docker does not filter (what host-gateway relies on).
297const BridgeInterface = "docker0"
298
299// bridgeLookup returns the IPv4 address the relay listens on for the steps
300// that run in a container; an error means there is none.
301type bridgeLookup func() (net.IP, error)
302
303// docker0IPv4 is the first IPv4 address of BridgeInterface, read when the job
304// starts: the nested daemon picks it, and init starts the executor once the
305// daemon answers (init.go), so the bridge exists by then when there is one.
306func docker0IPv4() (net.IP, error) {
307 ifi, err := net.InterfaceByName(BridgeInterface)
308 if err != nil {
309 return nil, err
310 }
311 addrs, err := ifi.Addrs()
312 if err != nil {
313 return nil, err
314 }
315 for _, a := range addrs {
316 if n, ok := a.(*net.IPNet); ok {
317 if ip4 := n.IP.To4(); ip4 != nil {
318 return ip4, nil
319 }
320 }
321 }
322 return nil, fmt.Errorf("%s has no IPv4 address", BridgeInterface)
323}
324
325// cacheRelay is the runtime service of a job in the sandbox (the cache, the
326// results service or both, on one channel): the relays, the URLs they listen
327// on and the token of the job. All are empty without a service; containerURL
328// and bridge are empty when the steps that run in a container cannot reach it.
329type cacheRelay struct {
330 relay, bridge *Relay
331 url, containerURL string
332 token, mode string
333 results, cacheOff bool
334}
335
336func (c cacheRelay) close() {
337 if c.relay != nil {
338 _ = c.relay.Close()
339 }
340 if c.bridge != nil {
341 _ = c.bridge.Close()
342 }
343}
344
345// startCacheRelay starts the relay of the runtime service for a job that has
346// one: it listens on the loopback port the host says and pipes every
347// connection to a new stream on the cache channel of the host (a sibling of
348// the executor's target). A relay that cannot start leaves the job without the
349// service, like a scope the host refuses: the job runs. A second relay listens
350// on the same port at the address bridge returns, for the job container and
351// the docker actions, whose loopback is their own; without it those steps get
352// no runtime variables.
353func startCacheRelay(ctx context.Context, target Target, spec *execpb.JobSpec, bridge bridgeLookup) cacheRelay {
354 port := spec.GetCachePort()
355 url := cacheURL(port)
356 if url == "" {
357 return cacheRelay{}
358 }
359 log := zerowrap.FromCtx(ctx)
360 dial := target.CacheTarget().DialStream
361 addr := net.JoinHostPort(CacheListenHost, strconv.FormatUint(uint64(port), 10))
362 r, err := StartRelay(ctx, addr, dial, DefaultRelayMax)
363 if err != nil {
364 log.Warn().Err(err).Msg("job runs without the cache")
365 return cacheRelay{}
366 }
367 results := spec.GetResults()
368 // A registry-only job (no cache scope, no results service: the host sends
369 // no token) needs dockerd's loopback relay, but no action runtime URLs or
370 // token: neither node nor docker actions get ACTIONS_* variables.
371 if spec.GetCacheToken() == "" || (!results && spec.GetCacheMode() == "") {
372 return cacheRelay{relay: r}
373 }
374 c := cacheRelay{relay: r, url: url, token: spec.GetCacheToken(), mode: spec.GetCacheMode(),
375 results: results, cacheOff: results && spec.GetCacheMode() == ""}
376 if bridge == nil {
377 return c
378 }
379 ip, err := bridge()
380 if err != nil {
381 log.Debug().Err(err).Msg("no docker bridge: steps in containers run without the runtime service")
382 return c
383 }
384 host := ip.String()
385 br, err := StartRelay(ctx, net.JoinHostPort(host, strconv.FormatUint(uint64(port), 10)), dial, DefaultRelayMax)
386 if err != nil {
387 log.Warn().Err(err).Msg("steps in containers run without the runtime service")
388 return c
389 }
390 c.bridge, c.containerURL = br, relayURL(host, port)
391 return c
392}
393
394// failSetup ends a job that cannot start, like a failed "Set up job": step 1
395// starts, logs the reason as an error, and fails; the job result is a failure.
396func failSetup(conn *execstream.Conn, msg string) int {
397 rep := &frameReporter{conn: conn, failed: make(chan struct{})}
398 now := time.Now()
399 rep.StepStarted(1, executor.SetUpJobName, now)
400 rep.Log(1, now, "##[error]"+msg)
401 rep.StepCompleted(1, executor.SetUpJobName, executor.ConclusionFailure, time.Now())
402 rep.JobCompleted(executor.Result{Conclusion: executor.ConclusionFailure, Error: msg})
403 return 1
404}
405
406// tokenPermissions is the permissions of GITHUB_TOKEN the server sent, nil when
407// it sent none (the executor then works them out from the workflow).
408func tokenPermissions(spec *execpb.JobSpec) map[string]string {
409 p := spec.GetTokenPermissions()
410 if p == nil {
411 return nil
412 }
413 if p.GetLevels() == nil {
414 return map[string]string{}
415 }
416 return p.GetLevels()
417}