forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
11.2 KiB
1//! The builtin executor through the real sandbox helper. The image source is a stub that returns
2//! a rootfs built from host binaries, so no registry is needed. The tests skip, with a message,
3//! when unprivileged user namespaces are unavailable.
4
5use std::collections::BTreeMap;
6use std::fs;
7use std::future::Future;
8use std::path::{Path, PathBuf};
9use std::process::Command;
10use std::time::{Duration, Instant};
11
12use forge_ci::proto::{JobAssignment, Outcome};
13use forge_runner::check::build_rootfs;
14use forge_runner::config::Limits;
15use forge_runner::executor::{BuiltinExecutor, ExecRequest, ExecResult, Executor, ImageSource};
16use forge_runner::oci::{ImageError, PreparedImage, RegistryAuth};
17use forge_runner::sandbox::job_cgroup_path;
18use nix::sys::signal::{Signal, kill};
19use nix::unistd::Pid;
20use tokio::sync::mpsc;
21use tokio_util::sync::CancellationToken;
22
23/// Hands out a fixed rootfs for every image, with an env that the job can observe.
24struct LocalRootfs(PathBuf);
25
26impl ImageSource for LocalRootfs {
27 fn prepare<'a>(
28 &'a self,
29 _image: &'a str,
30 _auth: Option<&'a RegistryAuth>,
31 _log: &'a mpsc::Sender<Vec<u8>>,
32 ) -> impl Future<Output = Result<PreparedImage, ImageError>> + Send + 'a {
33 std::future::ready(Ok(PreparedImage {
34 rootfs: self.0.clone(),
35 digest: format!("sha256:{}", "0".repeat(64)),
36 env: vec![
37 "PATH=/usr/local/bin:/usr/bin:/bin".to_owned(),
38 "FROM_IMAGE=yes".to_owned(),
39 ],
40 working_dir: None,
41 user: None,
42 }))
43 }
44}
45
46fn userns_available() -> bool {
47 let max = fs::read_to_string("/proc/sys/user/max_user_namespaces")
48 .ok()
49 .and_then(|s| s.trim().parse::<u64>().ok())
50 .unwrap_or(0);
51 if max == 0 {
52 return false;
53 }
54 Command::new("unshare")
55 .args(["--user", "--map-root-user", "true"])
56 .status()
57 .is_ok_and(|s| s.success())
58}
59
60/// Skips the test when user namespaces are unavailable. Returns `true` when the test should run.
61fn can_run() -> bool {
62 if userns_available() {
63 true
64 } else {
65 eprintln!("skipping: unprivileged user namespaces are not available on this host");
66 false
67 }
68}
69
70fn job(run: &str) -> JobAssignment {
71 JobAssignment {
72 job_id: "job_01".to_owned(),
73 job_token: "tok_01_secret_token".to_owned(),
74 repo: "owner/name".to_owned(),
75 clone_url: "https://forge.example/owner/name.git".to_owned(),
76 commit: "a".repeat(40),
77 git_ref: "refs/heads/main".to_owned(),
78 job_name: "build".to_owned(),
79 image: "docker.io/library/alpine:3".to_owned(),
80 run: run.to_owned(),
81 env: BTreeMap::from([("PLAIN".to_owned(), "visible".to_owned())]),
82 secrets: BTreeMap::from([("MY_TOKEN".to_owned(), "hunter2-secret".to_owned())]),
83 artifacts: Vec::new(),
84 timeout_minutes: 5,
85 network: false,
86 }
87}
88
89/// An executor over a rootfs in `dir`, and a fresh workspace under `dir`.
90fn executor(dir: &Path) -> (BuiltinExecutor<LocalRootfs>, PathBuf) {
91 let rootfs = dir.join("rootfs");
92 build_rootfs(&rootfs, &["sh", "sleep", "id", "touch"]).expect("build rootfs");
93 let workspace = dir.join("workspace");
94 fs::create_dir_all(&workspace).expect("create workspace");
95 let helper = PathBuf::from(env!("CARGO_BIN_EXE_forge-runner"));
96 let exec = BuiltinExecutor::new(
97 LocalRootfs(rootfs),
98 Limits::default(),
99 helper,
100 dir.join("jobs"),
101 );
102 (exec, workspace)
103}
104
105/// Runs `job` and returns the result with everything it logged.
106async fn run_job(
107 exec: &BuiltinExecutor<LocalRootfs>,
108 job: &JobAssignment,
109 workspace: &Path,
110 cancel: CancellationToken,
111) -> (ExecResult, String) {
112 let (tx, mut rx) = mpsc::channel::<Vec<u8>>(256);
113 let collector = tokio::spawn(async move {
114 let mut all = Vec::new();
115 while let Some(chunk) = rx.recv().await {
116 all.extend_from_slice(&chunk);
117 }
118 all
119 });
120 let result = exec
121 .run(ExecRequest {
122 job,
123 workspace,
124 logs: tx,
125 cancel,
126 })
127 .await;
128 let logs = collector.await.expect("log collector");
129 (result, String::from_utf8_lossy(&logs).into_owned())
130}
131
132#[tokio::test]
133async fn job_runs_in_the_sandbox_and_its_output_comes_back() {
134 let _serial = serial().await;
135 if !can_run() {
136 return;
137 }
138 let dir = tempfile::tempdir().expect("tempdir");
139 let (exec, workspace) = executor(dir.path());
140 let run = "echo hi-from-builtin; echo \"env=$FROM_IMAGE,$PLAIN,$CI\"; \
141 echo \"secret=$MY_TOKEN\"; id -u; touch /workspace/out.txt";
142 let (result, logs) = run_job(&exec, &job(run), &workspace, CancellationToken::new()).await;
143
144 assert_eq!(result.outcome, Outcome::Success, "{result:?}\n{logs}");
145 assert_eq!(result.exit_code, Some(0));
146 assert!(logs.contains("hi-from-builtin"), "{logs}");
147 assert!(logs.contains("env=yes,visible,true"), "{logs}");
148 assert!(logs.contains("secret=hunter2-secret"), "{logs}");
149 assert!(
150 logs.lines().any(|l| l == "0"),
151 "not root in the job: {logs}"
152 );
153 assert!(logs.contains("image sha256:"), "{logs}");
154 assert!(
155 workspace.join("out.txt").exists(),
156 "workspace write missing"
157 );
158 assert!(!dir.path().join("jobs/job_01").exists(), "job dir was kept");
159}
160
161#[tokio::test]
162async fn nonzero_exit_is_a_failure_with_its_code() {
163 let _serial = serial().await;
164 if !can_run() {
165 return;
166 }
167 let dir = tempfile::tempdir().expect("tempdir");
168 let (exec, workspace) = executor(dir.path());
169 let (result, logs) = run_job(&exec, &job("exit 3"), &workspace, CancellationToken::new()).await;
170 assert_eq!(result.outcome, Outcome::Failure, "{logs}");
171 assert_eq!(result.exit_code, Some(3));
172}
173
174#[tokio::test]
175async fn cancel_stops_the_container() {
176 let _serial = serial().await;
177 if !can_run() {
178 return;
179 }
180 let dir = tempfile::tempdir().expect("tempdir");
181 let (exec, workspace) = executor(dir.path());
182 let cancel = CancellationToken::new();
183 let stop = cancel.clone();
184 tokio::spawn(async move {
185 tokio::time::sleep(Duration::from_millis(1500)).await;
186 stop.cancel();
187 });
188 let started = Instant::now();
189 let (result, logs) = run_job(&exec, &job("sleep 120"), &workspace, cancel).await;
190 assert_eq!(result.outcome, Outcome::Cancelled, "{logs}");
191 // SIGTERM makes the helper kill its container, so this does not wait out the grace period.
192 assert!(
193 started.elapsed() < Duration::from_secs(25),
194 "{:?}",
195 started.elapsed()
196 );
197}
198
199#[tokio::test]
200async fn sandbox_setup_failure_is_an_error_with_the_reason() {
201 let _serial = serial().await;
202 if !can_run() {
203 return;
204 }
205 let dir = tempfile::tempdir().expect("tempdir");
206 let (exec, _workspace) = executor(dir.path());
207 let missing = dir.path().join("no-such-workspace");
208 let (result, logs) = run_job(&exec, &job("true"), &missing, CancellationToken::new()).await;
209 assert_eq!(result.outcome, Outcome::Error, "{logs}");
210 assert_eq!(result.exit_code, Some(125));
211 let message = result.message.unwrap_or_default();
212 assert!(
213 message.contains("workspace is not a directory"),
214 "{message}"
215 );
216}
217
218/// The builtin tests run in one process. Each takes this lock, so the helper-finding test below
219/// never sees another test's helper as its own child.
220static SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
221
222async fn serial() -> tokio::sync::MutexGuard<'static, ()> {
223 SERIAL.lock().await
224}
225
226/// The pids of this process's children that are sandbox helpers.
227fn helper_children() -> Vec<u32> {
228 let me = std::process::id();
229 let Ok(entries) = fs::read_dir("/proc") else {
230 return Vec::new();
231 };
232 entries
233 .filter_map(|e| e.ok()?.file_name().to_str()?.parse::<u32>().ok())
234 .filter(|&pid| {
235 let stat = fs::read_to_string(format!("/proc/{pid}/stat")).unwrap_or_default();
236 // The fields after the command name in parentheses start with the state, then ppid.
237 let ppid = stat
238 .rsplit_once(')')
239 .and_then(|(_, rest)| rest.split_whitespace().nth(1))
240 .and_then(|p| p.parse::<u32>().ok());
241 let cmdline = fs::read(format!("/proc/{pid}/cmdline")).unwrap_or_default();
242 ppid == Some(me) && cmdline.windows(9).any(|w| w == b"__sandbox".as_slice())
243 })
244 .collect()
245}
246
247/// The job cgroup directory of the helper with pid `helper`, for job `job_01`.
248fn job_cgroup_dir(helper: u32) -> PathBuf {
249 let rel = job_cgroup_path("job_01", helper).expect("job cgroup path");
250 Path::new("/sys/fs/cgroup").join(rel.strip_prefix("/").unwrap_or(&rel))
251}
252
253/// The pids listed in a cgroup's `cgroup.procs`.
254fn procs_in(dir: &Path) -> Vec<u32> {
255 fs::read_to_string(dir.join("cgroup.procs"))
256 .unwrap_or_default()
257 .lines()
258 .filter_map(|l| l.trim().parse().ok())
259 .collect()
260}
261
262/// Whether `pid` has exited. A zombie has exited too, it only waits to be reaped.
263fn process_gone(pid: u32) -> bool {
264 match fs::read_to_string(format!("/proc/{pid}/stat")) {
265 Err(_) => true,
266 Ok(stat) => {
267 stat.rsplit_once(')')
268 .and_then(|(_, rest)| rest.split_whitespace().next())
269 == Some("Z")
270 }
271 }
272}
273
274/// The helper is SIGKILLed while its job is still running, so it cannot tear the container down.
275/// The executor must then kill the job's processes, remove its cgroup, and remove its job dir.
276#[tokio::test]
277async fn sigkilled_helper_leaves_no_process_behind() {
278 let _serial = serial().await;
279 if !can_run() {
280 return;
281 }
282 let dir = tempfile::tempdir().expect("tempdir");
283 let (exec, workspace) = executor(dir.path());
284 // `trap '' TERM` makes the job ignore SIGTERM. Only the helper's SIGKILL path can stop it.
285 let run = "trap '' TERM; sleep 300";
286 let job_spec = job(run);
287
288 let killer = async {
289 let deadline = Instant::now() + Duration::from_secs(30);
290 let (helper, cgroup, procs) = loop {
291 assert!(Instant::now() < deadline, "the job never started");
292 let found = helper_children().into_iter().find_map(|pid| {
293 let cgroup = job_cgroup_dir(pid);
294 let procs = procs_in(&cgroup);
295 (!procs.is_empty()).then_some((pid, cgroup, procs))
296 });
297 if let Some(found) = found {
298 break found;
299 }
300 tokio::time::sleep(Duration::from_millis(50)).await;
301 };
302 kill(Pid::from_raw(helper as i32), Signal::SIGKILL).expect("kill the helper");
303 (cgroup, procs)
304 };
305
306 let (run_result, (cgroup, procs)) = tokio::join!(
307 run_job(&exec, &job_spec, &workspace, CancellationToken::new()),
308 killer
309 );
310 let (result, logs) = run_result;
311
312 assert_ne!(result.outcome, Outcome::Success, "{logs}");
313 assert!(
314 !cgroup.exists(),
315 "job cgroup {} is still there",
316 cgroup.display()
317 );
318 for pid in &procs {
319 assert!(process_gone(*pid), "job process {pid} survived the helper");
320 }
321 assert!(!dir.path().join("jobs/job_01").exists(), "job dir was kept");
322}