1// SPDX-License-Identifier: MIT
2
3package guest
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "io"
10 "os"
11 "os/exec"
12 "sync"
13 "syscall"
14 "time"
15
16 "bnema.dev/ermite-runner/internal/execstream"
17 "bnema.dev/ermite-runner/internal/execstream/execpb"
18)
19
20// DefaultScript is what the stub executor runs when the host sends no script.
21const DefaultScript = "uname -a; id; cat /etc/os-release; docker info"
22
23// Cancellation timings, like the official runner's ProcessInvoker: SIGINT, then
24// SIGTERM after 7.5 s, then SIGKILL 2.5 s later.
25const (
26 DefaultTermAfter = 7500 * time.Millisecond
27 DefaultKillAfter = 2500 * time.Millisecond
28 // DefaultWaitDelay bounds how long the executor waits for the output pipes
29 // to close after the script's shell exited (background children may hold
30 // them open).
31 DefaultWaitDelay = 500 * time.Millisecond
32)
33
34// ExecConfig configures the executor.
35type ExecConfig struct {
36 BootMs map[string]int64
37 Target Target
38 Sandbox string // "vm" or "container", reported in Hello
39 Version string
40 // Shell runs the script as `Shell -c script`; default /bin/sh.
41 Shell string
42 // StepUser is the account the steps of a job run as when the executor is
43 // root; default "runner".
44 StepUser string
45 // AllowRootSteps lets the steps run as root when StepUser does not exist.
46 // For test images only; set by ERMITE_RUNNER_ALLOW_ROOT_STEPS=1 in the
47 // executor's environment.
48 AllowRootSteps bool
49 // ConnectTimeout bounds the connection attempts (default 30 s).
50 ConnectTimeout time.Duration
51 // RunTimeout bounds the wait for the host's Run frame (default 60 s).
52 RunTimeout time.Duration
53 // TermAfter and KillAfter are the escalation delays after a Cancel (or after
54 // the executor's own context ends): SIGINT at once, SIGTERM after TermAfter,
55 // SIGKILL KillAfter later. WaitDelay is the pipe-drain grace after the shell
56 // exits. Zero means the defaults above.
57 TermAfter, KillAfter, WaitDelay time.Duration
58
59 // signal sends a signal to a process group; tests replace it to observe the
60 // escalation. Default: syscall.Kill(-pgid, sig).
61 signal func(pgid int, sig syscall.Signal)
62}
63
64func (c *ExecConfig) defaults() {
65 if c.Shell == "" {
66 c.Shell = "/bin/sh"
67 }
68 if c.StepUser == "" {
69 c.StepUser = "runner"
70 }
71 if c.ConnectTimeout == 0 {
72 c.ConnectTimeout = 30 * time.Second
73 }
74 if c.RunTimeout == 0 {
75 c.RunTimeout = 60 * time.Second
76 }
77 if c.TermAfter == 0 {
78 c.TermAfter = DefaultTermAfter
79 }
80 if c.KillAfter == 0 {
81 c.KillAfter = DefaultKillAfter
82 }
83 if c.WaitDelay == 0 {
84 c.WaitDelay = DefaultWaitDelay
85 }
86 if c.signal == nil {
87 c.signal = func(pgid int, sig syscall.Signal) { _ = syscall.Kill(-pgid, sig) }
88 }
89}
90
91// Exec is `ermite-runner exec`: it connects back to the host and receives one
92// Run. A Run with a job is executed by package executor (see runJob); a Run
93// without one is the stub of sandbox-check, which runs the script with the shell, streams stdout and
94// stderr as Output frames, and ends with an Exit frame. It returns the exit code
95// it reported. A Cancel frame, or the end of ctx, stops the script's process
96// group with escalating signals (see ExecConfig).
97func Exec(ctx context.Context, cfg ExecConfig) (int, error) {
98 cfg.defaults()
99 // The steps inherit the executor's umask: GitHub's 022, whatever started
100 // the executor (Init sets it too; a container's entrypoint may not).
101 syscall.Umask(StepUmask)
102 raw, err := cfg.Target.Dial(ctx, cfg.ConnectTimeout)
103 if err != nil {
104 return 1, err
105 }
106 defer func() { _ = raw.Close() }()
107 return serve(ctx, raw, cfg)
108}
109
110// frameWriter turns writes into Output frames of at most ChunkSize bytes. It is
111// the command's Stdout/Stderr, so no goroutine of ours sits between the pipe
112// and the stream.
113type frameWriter struct {
114 conn *execstream.Conn
115 kind execpb.StreamKind
116}
117
118func (w frameWriter) Write(p []byte) (int, error) {
119 n := 0
120 for len(p) > 0 {
121 chunk := p
122 if len(chunk) > execstream.ChunkSize {
123 chunk = chunk[:execstream.ChunkSize]
124 }
125 err := w.conn.Send(&execpb.Frame{Body: &execpb.Frame_Output{Output: &execpb.Output{
126 Stream: w.kind, Data: append([]byte(nil), chunk...), UnixNano: time.Now().UnixNano(),
127 }}})
128 if err != nil {
129 return n, err
130 }
131 n += len(chunk)
132 p = p[len(chunk):]
133 }
134 return n, nil
135}
136
137// escalator signals a process group and escalates after a cancel. Once finish
138// has been called (the shell was reaped) it never signals again: the group id
139// could have been reused by then.
140type escalator struct {
141 cfg ExecConfig
142 pgid int
143
144 mu sync.Mutex
145 finished bool
146 started bool
147 timers []*time.Timer
148}
149
150func (e *escalator) send(sig syscall.Signal) {
151 e.mu.Lock()
152 defer e.mu.Unlock()
153 if !e.finished {
154 e.cfg.signal(e.pgid, sig)
155 }
156}
157
158// cancel starts the escalation; only the first call does anything.
159func (e *escalator) cancel() {
160 e.mu.Lock()
161 if e.finished || e.started {
162 e.mu.Unlock()
163 return
164 }
165 e.started = true
166 e.cfg.signal(e.pgid, syscall.SIGINT)
167 e.timers = append(e.timers,
168 time.AfterFunc(e.cfg.TermAfter, func() { e.send(syscall.SIGTERM) }),
169 time.AfterFunc(e.cfg.TermAfter+e.cfg.KillAfter, func() { e.send(syscall.SIGKILL) }),
170 )
171 e.mu.Unlock()
172}
173
174// finish is called right after the shell was reaped: it kills what is left of
175// the group (background children), stops the timers and forbids further signals.
176func (e *escalator) finish() {
177 e.mu.Lock()
178 defer e.mu.Unlock()
179 if e.finished {
180 return
181 }
182 e.cfg.signal(e.pgid, syscall.SIGKILL)
183 e.finished = true
184 for _, t := range e.timers {
185 t.Stop()
186 }
187}
188
189func serve(ctx context.Context, rw io.ReadWriteCloser, cfg ExecConfig) (int, error) {
190 conn := execstream.NewConn(rw)
191 host, _ := os.Hostname()
192 if err := conn.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{
193 Version: cfg.Version, Sandbox: cfg.Sandbox, Hostname: host, BootMs: cfg.BootMs,
194 }}}); err != nil {
195 return 1, fmt.Errorf("send hello: %w", err)
196 }
197 run, vals, err := waitForRun(ctx, conn, cfg.RunTimeout)
198 if err != nil {
199 return 1, err
200 }
201 if run.GetJob() != nil {
202 return sendExit(conn, runJob(ctx, conn, cfg, run.GetJob(), vals), nil)
203 }
204 return runStub(ctx, conn, cfg, run)
205}
206
207// jobValues are the secrets and variables of the job that the host sent in
208// NamedValue frames before Run. They stay in memory.
209type jobValues struct{ secrets, vars map[string]string }
210
211// maxJobValues bounds what a host can make the executor keep before Run.
212const maxJobValues = 64 << 20
213
214// waitForRun waits for the Run frame of the host, bounded: a host that never
215// answers must not pin the sandbox. The NamedValue frames that come before it
216// are collected.
217func waitForRun(ctx context.Context, conn *execstream.Conn, timeout time.Duration) (*execpb.Run, jobValues, error) {
218 type recv struct {
219 run *execpb.Run
220 vals jobValues
221 err error
222 }
223 first := make(chan recv, 1)
224 go func() {
225 vals := jobValues{secrets: map[string]string{}, vars: map[string]string{}}
226 size := 0
227 for {
228 f, err := conn.Recv()
229 if err != nil {
230 first <- recv{err: fmt.Errorf("waiting for run: %w", err)}
231 return
232 }
233 switch b := f.Body.(type) {
234 case *execpb.Frame_Run:
235 first <- recv{run: b.Run, vals: vals}
236 return
237 case *execpb.Frame_Secret:
238 size += len(b.Secret.GetName()) + len(b.Secret.GetValue())
239 vals.secrets[b.Secret.GetName()] = b.Secret.GetValue()
240 case *execpb.Frame_Var:
241 size += len(b.Var.GetName()) + len(b.Var.GetValue())
242 vals.vars[b.Var.GetName()] = b.Var.GetValue()
243 default:
244 first <- recv{err: errors.New("first frame from host is not run")}
245 return
246 }
247 if size > maxJobValues {
248 first <- recv{err: errors.New("the host sent too many secrets and variables before run")}
249 return
250 }
251 }
252 }()
253 timer := time.NewTimer(timeout)
254 defer timer.Stop()
255 select {
256 case r := <-first:
257 return r.run, r.vals, r.err
258 case <-timer.C:
259 return nil, jobValues{}, errors.New("timed out waiting for the host's run frame")
260 case <-ctx.Done():
261 return nil, jobValues{}, ctx.Err()
262 }
263}
264
265// runStub is the stub executor of sandbox-check: it runs the script of a Run
266// without a job with the shell and streams its output.
267func runStub(ctx context.Context, conn *execstream.Conn, cfg ExecConfig, run *execpb.Run) (int, error) {
268 script := run.GetScript()
269 if script == "" {
270 script = DefaultScript
271 }
272 cmd := exec.Command(cfg.Shell, "-c", script)
273 cmd.Env = os.Environ()
274 for k, v := range run.GetEnv() {
275 cmd.Env = append(cmd.Env, k+"="+v)
276 }
277 cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
278 // Frame writers as stdout/stderr: exec copies the pipes for us, and
279 // WaitDelay bounds the wait when a background child keeps a pipe open.
280 cmd.Stdout = frameWriter{conn, execpb.StreamKind_STREAM_KIND_STDOUT}
281 cmd.Stderr = frameWriter{conn, execpb.StreamKind_STREAM_KIND_STDERR}
282 cmd.WaitDelay = cfg.WaitDelay
283 if err := cmd.Start(); err != nil {
284 return sendExit(conn, 127, fmt.Errorf("start %s: %w", cfg.Shell, err))
285 }
286 esc := &escalator{cfg: cfg, pgid: cmd.Process.Pid}
287
288 // Frames from the host while the script runs: a Cancel starts the escalation;
289 // a lost host kills the script rather than let it run unobserved.
290 go func() {
291 for {
292 f, err := conn.Recv()
293 if err != nil {
294 esc.send(syscall.SIGKILL)
295 return
296 }
297 if f.GetCancel() != nil {
298 esc.cancel()
299 }
300 }
301 }()
302 // The executor's own context (SIGTERM forwarded by init) cancels the same way.
303 stopCtx := context.AfterFunc(ctx, esc.cancel)
304 defer stopCtx()
305
306 waitErr := cmd.Wait()
307 esc.finish()
308 code, err := exitCode(waitErr)
309 return sendExit(conn, code, err)
310}
311
312// exitCode is the exit code to report for the error of cmd.Wait: 0 for a clean
313// exit (also when a background child kept the pipes open past WaitDelay: the
314// group was killed by esc.finish), the code of the process (128+signal when it
315// was killed, like a shell), or 1 with the error for anything else.
316func exitCode(waitErr error) (int, error) {
317 var ee *exec.ExitError
318 switch {
319 case waitErr == nil, errors.Is(waitErr, exec.ErrWaitDelay):
320 return 0, nil
321 case errors.As(waitErr, &ee):
322 code := ee.ExitCode()
323 if code < 0 {
324 if ws, ok := ee.Sys().(syscall.WaitStatus); ok && ws.Signaled() {
325 return 128 + int(ws.Signal()), nil
326 }
327 return 1, nil
328 }
329 return code, nil
330 }
331 return 1, waitErr
332}
333
334func sendExit(c *execstream.Conn, code int, cause error) (int, error) {
335 e := &execpb.Exit{Code: int32(code)}
336 if cause != nil {
337 e.Error = cause.Error()
338 }
339 if err := c.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: e}}); err != nil {
340 return code, fmt.Errorf("send exit: %w", err)
341 }
342 return code, nil
343}