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