forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
31.4 KiB
1//! Running a job's `run` script.
2//!
3//! `BuiltinExecutor` is the default: the image is prepared by an [`ImageSource`] (the
4//! [`ImageStore`] in production), and the script runs in a sandbox started by a helper process
5//! (`forge-runner __sandbox`). No external tool is needed.
6//!
7//! `PodmanExecutor` is the optional path: a rootless podman container with the workspace
8//! mounted at `/workspace`. Secret values and job env travel in the child process environment,
9//! and podman receives only `-e NAME`, so no value ever appears in argv.
10//!
11//! `HostExecutor` runs the script directly on the host. It exists only for tests and is compiled
12//! only under `cfg(test)` or the `test-host-executor` feature.
13
14use std::fs;
15use std::future::Future;
16use std::path::{Path, PathBuf};
17use std::pin::Pin;
18use std::process::{ExitStatus, Stdio};
19use std::time::Duration;
20
21use forge_ci::proto::{JobAssignment, Outcome};
22use nix::sys::signal::{Signal, kill};
23use nix::unistd::Pid;
24use oci_client::Reference;
25use tokio::io::AsyncWriteExt;
26use tokio::process::{Child, Command};
27use tokio::sync::mpsc;
28use tokio::task::JoinHandle;
29use tokio_util::sync::CancellationToken;
30
31use crate::config::{Limits, RegistryConfig};
32use crate::logs::forward_lines;
33use crate::oci::{ImageError, ImageStore, PreparedImage, RegistryAuth};
34use crate::sandbox::helper::EXIT_SETUP_FAILED;
35use crate::sandbox::idmap::{IdMap, remove_tree};
36use crate::sandbox::{SandboxRequest, job_cgroup_path};
37use crate::workspace::validate_job_id;
38
39/// The prefix of the sandbox helper's own error lines on stderr.
40const SANDBOX_PREFIX: &[u8] = b"forge-sandbox: ";
41
42/// How long to wait for the output readers to drain after the child has exited.
43const READER_GRACE: Duration = Duration::from_secs(10);
44/// How long to wait for the child to exit after it has been told to stop.
45const STOP_GRACE: Duration = Duration::from_secs(30);
46
47/// Everything an executor needs to run one job.
48pub struct ExecRequest<'a> {
49 pub job: &'a JobAssignment,
50 /// The checked-out source tree. The script runs with it as `/workspace` or as its cwd.
51 pub workspace: &'a Path,
52 /// Where merged stdout and stderr go, one line per message.
53 pub logs: mpsc::Sender<Vec<u8>>,
54 /// Cancelling this stops the job. The executor then reports `Outcome::Cancelled`.
55 pub cancel: CancellationToken,
56}
57
58/// The executor's answer. The runner adds timeouts and artifact results on top of it.
59#[derive(Clone, Debug, PartialEq, Eq)]
60pub struct ExecResult {
61 pub outcome: Outcome,
62 pub exit_code: Option<i32>,
63 pub message: Option<String>,
64}
65
66/// The boxed future an executor returns. Written out so `Executor` stays object safe.
67pub type ExecFuture<'a> = Pin<Box<dyn Future<Output = ExecResult> + Send + 'a>>;
68
69/// Runs jobs. The runner holds one as `Arc<dyn Executor>`.
70pub trait Executor: Send + Sync {
71 fn run<'a>(&'a self, req: ExecRequest<'a>) -> ExecFuture<'a>;
72}
73
74/// An executor failure, reported as `Outcome::Error`.
75#[derive(Debug, thiserror::Error)]
76pub enum ExecError {
77 #[error("invalid job: {0}")]
78 Invalid(String),
79}
80
81/// Env names go to podman as `-e NAME`, so they must be plain identifiers. A name with `=` in it
82/// would carry a value into argv.
83pub fn validate_env_name(name: &str) -> Result<(), ExecError> {
84 let mut chars = name.chars();
85 let first_ok = matches!(chars.next(), Some(c) if c.is_ascii_alphabetic() || c == '_');
86 let rest_ok = chars.all(|c| c.is_ascii_alphanumeric() || c == '_');
87 if first_ok && rest_ok && name.len() <= 256 {
88 Ok(())
89 } else {
90 Err(ExecError::Invalid(format!("bad env name {name:?}")))
91 }
92}
93
94/// The job's env and secrets as name/value pairs. Secrets win over plain env on a clash.
95pub fn job_env(job: &JobAssignment) -> Result<Vec<(String, String)>, ExecError> {
96 let mut pairs: Vec<(String, String)> = Vec::new();
97 for (k, v) in job.env.iter().chain(job.secrets.iter()) {
98 validate_env_name(k)?;
99 pairs.retain(|(name, _)| name != k);
100 pairs.push((k.clone(), v.clone()));
101 }
102 Ok(pairs)
103}
104
105/// The podman invocation for a job: argv (no values in it) and the env the child needs.
106#[derive(Debug, PartialEq, Eq)]
107pub struct PodmanInvocation {
108 pub args: Vec<String>,
109 pub env: Vec<(String, String)>,
110 pub container: String,
111}
112
113fn check_image(image: &str) -> Result<(), ExecError> {
114 // Podman would read a leading `-` as an option, and whitespace is never valid in a reference.
115 if image.is_empty() || image.starts_with('-') || image.contains(char::is_whitespace) {
116 return Err(ExecError::Invalid(format!("bad image {image:?}")));
117 }
118 Ok(())
119}
120
121fn check_limit(value: &str, what: &str) -> Result<(), ExecError> {
122 let ok = !value.is_empty()
123 && !value.starts_with('-')
124 && value.chars().all(|c| c.is_ascii_alphanumeric() || c == '.');
125 if ok {
126 Ok(())
127 } else {
128 Err(ExecError::Invalid(format!("bad {what} limit {value:?}")))
129 }
130}
131
132/// Builds the podman invocation for `job`. Pure, so the argv can be tested.
133pub fn podman_invocation(
134 job: &JobAssignment,
135 workspace: &Path,
136 limits: &Limits,
137) -> Result<PodmanInvocation, ExecError> {
138 validate_job_id(&job.job_id).map_err(|e| ExecError::Invalid(e.to_string()))?;
139 check_image(&job.image)?;
140 check_limit(&limits.memory, "memory")?;
141 check_limit(&limits.cpus, "cpus")?;
142 let ws = workspace
143 .to_str()
144 .filter(|s| !s.contains(':') && !s.contains(','))
145 .ok_or_else(|| ExecError::Invalid("workspace path has a ':' or ','".into()))?;
146
147 let env = job_env(job)?;
148 let container = format!("forge-job-{}", job.job_id);
149 let network = if job.network { "slirp4netns" } else { "none" };
150
151 let mut args: Vec<String> = [
152 "run",
153 "--rm",
154 "--name",
155 &container,
156 "--pull=newer",
157 &format!("--network={network}"),
158 "--memory",
159 &limits.memory,
160 "--cpus",
161 &limits.cpus,
162 "--pids-limit",
163 ]
164 .iter()
165 .map(|s| (*s).to_owned())
166 .collect();
167 args.push(limits.pids.to_string());
168 for flag in [
169 "--security-opt",
170 "no-new-privileges",
171 "--cap-drop",
172 "ALL",
173 "-v",
174 ] {
175 args.push(flag.to_owned());
176 }
177 args.push(format!("{ws}:/workspace:Z"));
178 args.extend(["-w".to_owned(), "/workspace".to_owned()]);
179 for (name, _) in &env {
180 args.push("-e".to_owned());
181 args.push(name.clone());
182 }
183 args.push(job.image.clone());
184 args.extend([
185 "sh".to_owned(),
186 "-eu".to_owned(),
187 "-c".to_owned(),
188 job.run.clone(),
189 ]);
190
191 Ok(PodmanInvocation {
192 args,
193 env,
194 container,
195 })
196}
197
198/// How a finished child maps to a job result.
199fn classify(status: ExitStatus) -> ExecResult {
200 match status.code() {
201 Some(0) => ExecResult {
202 outcome: Outcome::Success,
203 exit_code: Some(0),
204 message: None,
205 },
206 // podman itself exits 125 when it cannot run the container (no image, no storage, ...).
207 Some(125) => ExecResult {
208 outcome: Outcome::Error,
209 exit_code: Some(125),
210 message: Some("podman failed to run the container (exit 125)".to_owned()),
211 },
212 Some(code) => ExecResult {
213 outcome: Outcome::Failure,
214 exit_code: Some(code),
215 message: None,
216 },
217 None => ExecResult {
218 outcome: Outcome::Failure,
219 exit_code: None,
220 message: Some("job was terminated by a signal".to_owned()),
221 },
222 }
223}
224
225fn error_result(message: impl Into<String>) -> ExecResult {
226 ExecResult {
227 outcome: Outcome::Error,
228 exit_code: None,
229 message: Some(message.into()),
230 }
231}
232
233/// How to stop a job when it is cancelled.
234enum Stopper {
235 Podman {
236 container: String,
237 },
238 #[cfg(any(test, feature = "test-host-executor"))]
239 ProcessGroup {
240 pid: u32,
241 },
242 /// The sandbox helper. It is sent SIGTERM and stops its own container, see [`stop_child`].
243 Helper {
244 pid: u32,
245 },
246}
247
248impl Stopper {
249 async fn stop(&self) {
250 match self {
251 Stopper::Podman { container } => {
252 let quiet = |cmd: &mut Command| {
253 cmd.stdin(Stdio::null())
254 .stdout(Stdio::null())
255 .stderr(Stdio::null());
256 };
257 let mut kill = Command::new("podman");
258 kill.args(["kill", container.as_str()]);
259 quiet(&mut kill);
260 let _ = kill.status().await;
261 let mut rm = Command::new("podman");
262 rm.args(["rm", "-f", container.as_str()]);
263 quiet(&mut rm);
264 let _ = rm.status().await;
265 }
266 #[cfg(any(test, feature = "test-host-executor"))]
267 Stopper::ProcessGroup { pid } => {
268 let mut kill = Command::new("kill");
269 kill.args(["-KILL", "--", &format!("-{pid}")])
270 .stdin(Stdio::null())
271 .stdout(Stdio::null())
272 .stderr(Stdio::null());
273 let _ = kill.status().await;
274 }
275 Stopper::Helper { .. } => {}
276 }
277 }
278}
279
280/// Stops the child on cancellation and returns its exit status.
281///
282/// The sandbox helper gets SIGTERM and gets to kill and delete its container itself. SIGKILL
283/// follows only if it does not exit within [`STOP_GRACE`]. Other children are killed at once.
284async fn stop_child(child: &mut Child, stopper: Stopper) -> std::io::Result<ExitStatus> {
285 if let Stopper::Helper { pid } = stopper {
286 let _ = kill(Pid::from_raw(pid as i32), Signal::SIGTERM);
287 return match tokio::time::timeout(STOP_GRACE, child.wait()).await {
288 Ok(status) => status,
289 Err(_) => {
290 let _ = child.start_kill();
291 child.wait().await
292 }
293 };
294 }
295 stopper.stop().await;
296 let _ = child.start_kill();
297 tokio::time::timeout(STOP_GRACE, child.wait())
298 .await
299 .unwrap_or_else(|_| Err(std::io::Error::other("child did not stop")))
300}
301
302/// What [`supervise`] saw when the child finished.
303struct Supervised {
304 cancelled: bool,
305 status: std::io::Result<ExitStatus>,
306 /// The sandbox helper's last `forge-sandbox: <reason>` line on stderr, when it printed one.
307 sandbox_error: Option<String>,
308}
309
310/// Copies stderr lines to the log and remembers the last sandbox error line.
311async fn tap_sandbox_errors(
312 mut lines: mpsc::Receiver<Vec<u8>>,
313 logs: mpsc::Sender<Vec<u8>>,
314) -> Option<String> {
315 let mut found = None;
316 while let Some(chunk) = lines.recv().await {
317 let is_error = chunk.starts_with(SANDBOX_PREFIX)
318 && !chunk[SANDBOX_PREFIX.len()..].starts_with(b"warning:");
319 if is_error {
320 found = Some(String::from_utf8_lossy(&chunk).trim_end().to_owned());
321 }
322 if logs.send(chunk).await.is_err() {
323 break;
324 }
325 }
326 found
327}
328
329/// Forwards the child's output, waits for it, and reports how it ended. On cancellation the
330/// child is stopped first. `cleanup` runs as soon as the child has exited, before the output is
331/// drained: a container keeps the output pipes open until it is killed.
332async fn supervise(
333 mut child: Child,
334 logs: mpsc::Sender<Vec<u8>>,
335 cancel: CancellationToken,
336 stopper: Stopper,
337 cleanup: Option<JobCleanup>,
338) -> Supervised {
339 let mut readers: Vec<JoinHandle<()>> = Vec::new();
340 if let Some(out) = child.stdout.take() {
341 readers.push(tokio::spawn(forward_lines(out, logs.clone())));
342 }
343 let mut tap = None;
344 if let Some(err) = child.stderr.take() {
345 let (tx, rx) = mpsc::channel(64);
346 readers.push(tokio::spawn(forward_lines(err, tx)));
347 tap = Some(tokio::spawn(tap_sandbox_errors(rx, logs.clone())));
348 }
349 drop(logs);
350
351 // The wait future must be dropped before the stop path borrows the child again.
352 let early = tokio::select! {
353 status = child.wait() => Some(status),
354 _ = cancel.cancelled() => None,
355 };
356 let (status, cancelled) = match early {
357 Some(status) => (status, false),
358 None => (stop_child(&mut child, stopper).await, true),
359 };
360 if let Some(job) = cleanup {
361 reap_job(&job).await;
362 }
363
364 let drain = async {
365 for reader in readers.iter_mut() {
366 let _ = reader.await;
367 }
368 };
369 if tokio::time::timeout(READER_GRACE, drain).await.is_err() {
370 // A grandchild still holds the pipe. Stop reading rather than hang the job.
371 for reader in &readers {
372 reader.abort();
373 }
374 }
375 let sandbox_error = match tap {
376 Some(tap) => match tokio::time::timeout(READER_GRACE, tap).await {
377 Ok(Ok(found)) => found,
378 _ => None,
379 },
380 None => None,
381 };
382
383 Supervised {
384 cancelled,
385 status,
386 sandbox_error,
387 }
388}
389
390/// The result for a child that was supervised with the podman or host rules.
391fn finish(supervised: Supervised) -> ExecResult {
392 if supervised.cancelled {
393 return cancelled_result();
394 }
395 match supervised.status {
396 Ok(status) => classify(status),
397 Err(e) => error_result(format!("waiting for the job failed: {e}")),
398 }
399}
400
401fn cancelled_result() -> ExecResult {
402 ExecResult {
403 outcome: Outcome::Cancelled,
404 exit_code: None,
405 message: Some("job was cancelled".to_owned()),
406 }
407}
408
409/// Runs jobs in rootless podman.
410#[derive(Clone, Debug)]
411pub struct PodmanExecutor {
412 limits: Limits,
413}
414
415impl PodmanExecutor {
416 pub fn new(limits: Limits) -> Self {
417 Self { limits }
418 }
419}
420
421impl Executor for PodmanExecutor {
422 fn run<'a>(&'a self, req: ExecRequest<'a>) -> ExecFuture<'a> {
423 Box::pin(async move {
424 let inv = match podman_invocation(req.job, req.workspace, &self.limits) {
425 Ok(inv) => inv,
426 Err(e) => return error_result(e.to_string()),
427 };
428 let mut cmd = Command::new("podman");
429 cmd.args(&inv.args)
430 .envs(inv.env.iter().map(|(k, v)| (k.as_str(), v.as_str())))
431 .stdin(Stdio::null())
432 .stdout(Stdio::piped())
433 .stderr(Stdio::piped())
434 .kill_on_drop(true);
435 let child = match cmd.spawn() {
436 Ok(child) => child,
437 Err(e) => return error_result(format!("cannot start podman: {e}")),
438 };
439 finish(
440 supervise(
441 child,
442 req.logs,
443 req.cancel,
444 Stopper::Podman {
445 container: inv.container,
446 },
447 None,
448 )
449 .await,
450 )
451 })
452 }
453}
454
455/// Runs the script on the host with no isolation. Tests only.
456#[cfg(any(test, feature = "test-host-executor"))]
457#[derive(Clone, Copy, Debug, Default)]
458pub struct HostExecutor;
459
460#[cfg(any(test, feature = "test-host-executor"))]
461impl Executor for HostExecutor {
462 fn run<'a>(&'a self, req: ExecRequest<'a>) -> ExecFuture<'a> {
463 Box::pin(async move {
464 let env = match job_env(req.job) {
465 Ok(env) => env,
466 Err(e) => return error_result(e.to_string()),
467 };
468 let mut cmd = Command::new("sh");
469 cmd.args(["-eu", "-c", req.job.run.as_str()])
470 .current_dir(req.workspace)
471 .envs(env.iter().map(|(k, v)| (k.as_str(), v.as_str())))
472 .stdin(Stdio::null())
473 .stdout(Stdio::piped())
474 .stderr(Stdio::piped())
475 .process_group(0)
476 .kill_on_drop(true);
477 let child = match cmd.spawn() {
478 Ok(child) => child,
479 Err(e) => return error_result(format!("cannot start sh: {e}")),
480 };
481 let Some(pid) = child.id() else {
482 return error_result("child exited before its pid was read");
483 };
484 finish(
485 supervise(
486 child,
487 req.logs,
488 req.cancel,
489 Stopper::ProcessGroup { pid },
490 None,
491 )
492 .await,
493 )
494 })
495 }
496}
497
498/// Where the builtin executor gets its rootfs. [`ImageStore`] pulls and unpacks images. Tests
499/// supply a local rootfs directory instead.
500pub trait ImageSource: Send + Sync {
501 /// Makes `image` available and returns its rootfs and config. Progress lines go to `log`.
502 fn prepare<'a>(
503 &'a self,
504 image: &'a str,
505 auth: Option<&'a RegistryAuth>,
506 log: &'a mpsc::Sender<Vec<u8>>,
507 ) -> impl Future<Output = Result<PreparedImage, ImageError>> + Send + 'a;
508}
509
510impl ImageSource for ImageStore {
511 fn prepare<'a>(
512 &'a self,
513 image: &'a str,
514 auth: Option<&'a RegistryAuth>,
515 log: &'a mpsc::Sender<Vec<u8>>,
516 ) -> impl Future<Output = Result<PreparedImage, ImageError>> + Send + 'a {
517 self.ensure(image, auth, log)
518 }
519}
520
521/// Runs jobs in a libcontainer sandbox, started as `helper` (`forge-runner __sandbox`). It needs
522/// no external container tool. See `docs/ci-builtin-executor.md`.
523pub struct BuiltinExecutor<I = ImageStore> {
524 images: I,
525 limits: Limits,
526 helper: PathBuf,
527 /// Holds one `<job id>` directory per running job, for the overlay and container state.
528 jobs_dir: PathBuf,
529 /// `[[registry]]` logins, as (host, login).
530 registries: Vec<(String, RegistryAuth)>,
531 /// The id map jobs run under. Single unless set by [`BuiltinExecutor::with_idmap`].
532 idmap: IdMap,
533}
534
535impl<I: ImageSource> BuiltinExecutor<I> {
536 pub fn new(images: I, limits: Limits, helper: PathBuf, jobs_dir: PathBuf) -> Self {
537 Self {
538 images,
539 limits,
540 helper,
541 jobs_dir,
542 registries: Vec::new(),
543 idmap: IdMap::single(),
544 }
545 }
546
547 /// Sets the id map that jobs run under. Job dirs are removed under it too.
548 pub fn with_idmap(mut self, map: IdMap) -> Self {
549 self.idmap = map;
550 self
551 }
552
553 /// Sets the `[[registry]]` logins. Each one is used for images on its host.
554 pub fn with_registries(mut self, registries: &[RegistryConfig]) -> Self {
555 self.registries = registries
556 .iter()
557 .map(|r| {
558 (
559 r.host.clone(),
560 RegistryAuth {
561 username: r.username.clone(),
562 password: r.password.clone(),
563 },
564 )
565 })
566 .collect();
567 self
568 }
569
570 /// The login for `image`, matched by its registry host. Docker Hub is `docker.io`.
571 fn auth_for(&self, image: &str) -> Option<&RegistryAuth> {
572 let reference: Reference = image.parse().ok()?;
573 let host = reference.registry();
574 self.registries
575 .iter()
576 .find(|(registry, _)| registry == host)
577 .map(|(_, auth)| auth)
578 }
579
580 async fn run_job(&self, req: ExecRequest<'_>) -> ExecResult {
581 let ExecRequest {
582 job,
583 workspace,
584 logs,
585 cancel,
586 } = req;
587 if let Err(e) = validate_job_id(&job.job_id) {
588 return error_result(e.to_string());
589 }
590 if let Err(e) = check_image(&job.image) {
591 return error_result(e.to_string());
592 }
593 let env = match job_env(job) {
594 Ok(env) => env,
595 Err(e) => return error_result(e.to_string()),
596 };
597
598 let auth = self.auth_for(&job.image);
599 let prepared = tokio::select! {
600 prepared = self.images.prepare(&job.image, auth, &logs) => prepared,
601 _ = cancel.cancelled() => return cancelled_result(),
602 };
603 let image = match prepared {
604 Ok(image) => image,
605 Err(e) => return error_result(format!("cannot prepare image {}: {e}", job.image)),
606 };
607 let _ = logs
608 .send(format!("image {}\n", image.digest).into_bytes())
609 .await;
610
611 let job_dir = self.jobs_dir.join(&job.job_id);
612 // A leftover from an earlier run of this job id would hold stale overlay state.
613 if let Err(e) = remove_job_dir(&self.idmap, &self.helper, &job_dir).await {
614 tracing::warn!("cannot remove stale {}: {e}", job_dir.display());
615 }
616 let request = SandboxRequest {
617 job_id: job.job_id.clone(),
618 rootfs: image.rootfs,
619 job_dir,
620 workspace: workspace.to_owned(),
621 image_env: image.env,
622 env,
623 run: job.run.clone(),
624 network: job.network,
625 limits: self.limits.clone(),
626 idmap: self.idmap.clone(),
627 };
628 // The request holds the secrets. It goes to the helper's stdin and nowhere else.
629 let json = match serde_json::to_vec(&request) {
630 Ok(json) => json,
631 Err(e) => return error_result(format!("cannot encode the sandbox request: {e}")),
632 };
633 drop(request);
634
635 let mut cmd = Command::new(&self.helper);
636 cmd.arg("__sandbox")
637 .stdin(Stdio::piped())
638 .stdout(Stdio::piped())
639 .stderr(Stdio::piped())
640 .kill_on_drop(true);
641 let mut child = match cmd.spawn() {
642 Ok(child) => child,
643 Err(e) => return error_result(format!("cannot start the sandbox helper: {e}")),
644 };
645 let Some(pid) = child.id() else {
646 return error_result("sandbox helper exited before its pid was read");
647 };
648 if let Some(mut stdin) = child.stdin.take() {
649 // The helper reads the whole request before it does anything else.
650 if let Err(e) = stdin.write_all(&json).await {
651 return error_result(format!("cannot send the job to the sandbox helper: {e}"));
652 }
653 }
654 drop(json);
655
656 let Supervised {
657 cancelled,
658 status,
659 sandbox_error,
660 } = supervise(
661 child,
662 logs,
663 cancel,
664 Stopper::Helper { pid },
665 Some(JobCleanup {
666 helper_pid: pid,
667 job_id: job.job_id.clone(),
668 job_dir: self.jobs_dir.join(&job.job_id),
669 idmap: self.idmap.clone(),
670 helper: self.helper.clone(),
671 }),
672 )
673 .await;
674 if cancelled {
675 return cancelled_result();
676 }
677 match status {
678 Err(e) => error_result(format!("waiting for the sandbox failed: {e}")),
679 Ok(status) => match status.code() {
680 Some(0) => ExecResult {
681 outcome: Outcome::Success,
682 exit_code: Some(0),
683 message: None,
684 },
685 // The helper's own setup failure. A job that exits 125 by itself is a failure.
686 Some(EXIT_SETUP_FAILED) if sandbox_error.is_some() => {
687 let line = sandbox_error.unwrap_or_default();
688 let reason = line.strip_prefix("forge-sandbox: ").unwrap_or(&line);
689 ExecResult {
690 outcome: Outcome::Error,
691 exit_code: Some(EXIT_SETUP_FAILED),
692 message: Some(format!("sandbox setup failed: {reason}")),
693 }
694 }
695 Some(code) => ExecResult {
696 outcome: Outcome::Failure,
697 exit_code: Some(code),
698 message: None,
699 },
700 None => ExecResult {
701 outcome: Outcome::Failure,
702 exit_code: None,
703 message: Some("job was terminated by a signal".to_owned()),
704 },
705 },
706 }
707 }
708}
709
710/// Where cgroup2 is mounted. Job cgroup paths are relative to it.
711const CGROUP_ROOT: &str = "/sys/fs/cgroup";
712/// How often, and how many times, to retry removing a job cgroup while its processes exit.
713const REAP_POLL: Duration = Duration::from_millis(100);
714const REAP_POLLS: u32 = 50;
715
716/// Kills every process in a job cgroup. `cgroup.kill` does it in one write (Linux 5.14 and up).
717/// Older kernels get a SIGKILL to each pid in `cgroup.procs`.
718fn kill_cgroup(dir: &Path) {
719 if fs::write(dir.join("cgroup.kill"), "1").is_ok() {
720 return;
721 }
722 let Ok(procs) = fs::read_to_string(dir.join("cgroup.procs")) else {
723 return;
724 };
725 for pid in procs
726 .lines()
727 .filter_map(|line| line.trim().parse::<i32>().ok())
728 {
729 let _ = kill(Pid::from_raw(pid), Signal::SIGKILL);
730 }
731}
732
733/// What to clean up once a job's sandbox helper has exited. See [`reap_job`].
734struct JobCleanup {
735 helper_pid: u32,
736 job_id: String,
737 job_dir: PathBuf,
738 idmap: IdMap,
739 helper: PathBuf,
740}
741
742/// Cleans up after a job's helper has exited, whatever the reason. A helper killed with SIGKILL
743/// cannot delete its container, so the container could otherwise keep running. This kills what is
744/// left in the job cgroup, removes the cgroup, and removes the job dir.
745async fn reap_job(job: &JobCleanup) {
746 if let Ok(relative) = job_cgroup_path(&job.job_id, job.helper_pid) {
747 let dir = Path::new(CGROUP_ROOT).join(relative.strip_prefix("/").unwrap_or(&relative));
748 if dir.exists() {
749 kill_cgroup(&dir);
750 // rmdir fails while processes remain, so retry until they have exited.
751 for _ in 0..REAP_POLLS {
752 if fs::remove_dir(&dir).is_ok() || !dir.exists() {
753 break;
754 }
755 tokio::time::sleep(REAP_POLL).await;
756 }
757 }
758 }
759 if let Err(e) = remove_job_dir(&job.idmap, &job.helper, &job.job_dir).await {
760 tracing::warn!("cannot remove {}: {e}", job.job_dir.display());
761 }
762}
763
764/// Removes a job dir under `map`. Its files may be owned by subordinate ids, so this goes through
765/// the helper in subuid mode (see [`remove_tree`]). The removal blocks, so it runs off the async
766/// runtime.
767async fn remove_job_dir(map: &IdMap, helper: &Path, job_dir: &Path) -> std::io::Result<()> {
768 let (map, helper, job_dir) = (map.clone(), helper.to_owned(), job_dir.to_owned());
769 tokio::task::spawn_blocking(move || remove_tree(&map, &job_dir, &helper))
770 .await
771 .map_err(std::io::Error::other)?
772}
773
774impl<I: ImageSource> Executor for BuiltinExecutor<I> {
775 fn run<'a>(&'a self, req: ExecRequest<'a>) -> ExecFuture<'a> {
776 Box::pin(self.run_job(req))
777 }
778}
779
780#[cfg(test)]
781mod tests {
782 use super::*;
783 use std::collections::BTreeMap;
784
785 const SECRET: &str = "hunter2-SUPER-SECRET-value";
786
787 pub(crate) fn fixture_job(run: &str) -> JobAssignment {
788 JobAssignment {
789 job_id: "job_01".to_owned(),
790 job_token: "tok_01_secret_token".to_owned(),
791 repo: "owner/name".to_owned(),
792 clone_url: "https://forge.example/owner/name.git".to_owned(),
793 commit: "a".repeat(40),
794 git_ref: "refs/heads/main".to_owned(),
795 job_name: "build".to_owned(),
796 image: "docker.io/library/alpine:3".to_owned(),
797 run: run.to_owned(),
798 env: BTreeMap::from([("PLAIN".to_owned(), "visible".to_owned())]),
799 secrets: BTreeMap::from([("MY_TOKEN".to_owned(), SECRET.to_owned())]),
800 artifacts: Vec::new(),
801 timeout_minutes: 5,
802 network: false,
803 }
804 }
805
806 #[test]
807 fn podman_argv_never_contains_a_secret_value() {
808 let job = fixture_job("echo $MY_TOKEN");
809 let inv = podman_invocation(&job, Path::new("/work/job_01"), &Limits::default())
810 .expect("invocation");
811 assert!(inv.args.iter().all(|a| !a.contains(SECRET)));
812 assert!(inv.args.iter().all(|a| !a.contains("visible")));
813 // The values travel in the child env, by name.
814 assert!(
815 inv.env
816 .contains(&("MY_TOKEN".to_owned(), SECRET.to_owned()))
817 );
818 assert!(
819 inv.env
820 .contains(&("PLAIN".to_owned(), "visible".to_owned()))
821 );
822 let names: Vec<&str> = inv
823 .args
824 .windows(2)
825 .filter(|w| w[0] == "-e")
826 .map(|w| w[1].as_str())
827 .collect();
828 assert_eq!(names, vec!["PLAIN", "MY_TOKEN"]);
829 }
830
831 #[test]
832 fn podman_argv_has_the_isolation_flags() {
833 let job = fixture_job("true");
834 let inv = podman_invocation(&job, Path::new("/work/job_01"), &Limits::default())
835 .expect("invocation");
836 let joined = inv.args.join(" ");
837 assert!(joined.starts_with("run --rm --name forge-job-job_01 --pull=newer"));
838 assert!(joined.contains("--network=none"));
839 assert!(joined.contains("--memory 4g --cpus 2 --pids-limit 4096"));
840 assert!(joined.contains("--security-opt no-new-privileges --cap-drop ALL"));
841 assert!(joined.contains("-v /work/job_01:/workspace:Z -w /workspace"));
842 assert!(joined.ends_with("docker.io/library/alpine:3 sh -eu -c true"));
843 }
844
845 #[test]
846 fn network_true_selects_slirp() {
847 let mut job = fixture_job("true");
848 job.network = true;
849 let inv = podman_invocation(&job, Path::new("/w"), &Limits::default()).expect("inv");
850 assert!(inv.args.contains(&"--network=slirp4netns".to_owned()));
851 }
852
853 #[test]
854 fn hostile_values_are_refused() {
855 let ws = Path::new("/w");
856 let mut job = fixture_job("true");
857 job.image = "--privileged".to_owned();
858 assert!(podman_invocation(&job, ws, &Limits::default()).is_err());
859
860 let mut job = fixture_job("true");
861 job.env.insert("A=B".to_owned(), "x".to_owned());
862 assert!(podman_invocation(&job, ws, &Limits::default()).is_err());
863
864 let mut job = fixture_job("true");
865 job.job_id = "../escape".to_owned();
866 assert!(podman_invocation(&job, ws, &Limits::default()).is_err());
867
868 let job = fixture_job("true");
869 assert!(podman_invocation(&job, Path::new("/w:evil"), &Limits::default()).is_err());
870 }
871
872 #[test]
873 fn exit_codes_map_to_outcomes() {
874 use std::os::unix::process::ExitStatusExt;
875 let ok = classify(ExitStatus::from_raw(0));
876 assert_eq!(ok.outcome, Outcome::Success);
877 let fail = classify(ExitStatus::from_raw(3 << 8));
878 assert_eq!(fail.outcome, Outcome::Failure);
879 assert_eq!(fail.exit_code, Some(3));
880 let podman = classify(ExitStatus::from_raw(125 << 8));
881 assert_eq!(podman.outcome, Outcome::Error);
882 }
883
884 /// Needs rootless podman and network access to pull the image.
885 #[tokio::test]
886 #[ignore = "requires rootless podman and registry access"]
887 async fn podman_runs_a_real_container() {
888 let dir = tempfile::tempdir().expect("tempdir");
889 let ws = dir.path().join("ws");
890 std::fs::create_dir_all(&ws).expect("mkdir");
891 let job = fixture_job(
892 "echo hi-from-container; echo \"s=$MY_TOKEN\" >&2; touch /workspace/out.txt",
893 );
894 let (tx, mut rx) = mpsc::channel(64);
895 let exec = PodmanExecutor::new(Limits::default());
896 let result = exec
897 .run(ExecRequest {
898 job: &job,
899 workspace: &ws,
900 logs: tx,
901 cancel: CancellationToken::new(),
902 })
903 .await;
904 assert_eq!(result.outcome, Outcome::Success, "{result:?}");
905 let mut all = Vec::new();
906 while let Some(chunk) = rx.recv().await {
907 all.extend_from_slice(&chunk);
908 }
909 let text = String::from_utf8_lossy(&all).into_owned();
910 assert!(text.contains("hi-from-container"));
911 assert!(ws.join("out.txt").exists());
912 }
913}