Skip to content
Sign up

flagon-io/g1t

public
0
936 linesCodeBlameRaw
1//! Inside the sandbox's VM: `g1t-runner sandbox guest …`, which the
2//! launcher on the computer runs (through `wsl.exe --exec` on Windows, as
3//! the VM's systemd service on macOS and Linux).
4//!
5//! `guest run` is the VM's supervisor. It
6//!
7//! 1. on WSL, takes away what WSL puts in every distribution that reaches
8//! Windows: `/mnt/wsl` (shared by every distribution, where Docker
9//! Desktop keeps sockets to Windows' Docker), WSLg's display and sound
10//! sockets, Windows' GPU driver folders, and `/dev/dxg`; and checks
11//! `/etc/wsl.conf` is the image's, which turns off drives and Windows
12//! programs ([`WSL_CONF`]);
13//! 2. refuses to go on if anything of the computer's is still mounted, or
14//! a container engine's socket other than its own is reachable
15//! ([`problems`]);
16//! 3. starts Sysbox and a Docker Engine of the VM's own, and adds rules
17//! that keep every container off the computer's own address (the VM's
18//! gateway) but for DNS ([`firewall_rules`]);
19//! 4. runs the runner's image under `--runtime=sysbox-runc` with the
20//! registration on `/var/lib/g1t-sandbox/state`, and keeps saying how
21//! it is in `/var/lib/g1t-sandbox/status.json` (and on the console);
22//! 5. on `guest stop` (or SIGTERM), asks the runner to drain or stop,
23//! waits for it, and stops everything.
24//!
25//! Root in this VM is not a wall on WSL: every WSL distribution, the
26//! person's own among them, runs in one VM, and root in any of them can
27//! reach the others' disks. That is why the runner runs under Sysbox here
28//! and its jobs in containers inside it: a job is two walls from this
29//! VM's root, and three from the computer.
30
31use std::net::Ipv4Addr;
32use std::path::Path;
33
34use anyhow::{Context, Result, bail};
35use serde::{Deserialize, Serialize};
36use serde_json::{Value, json};
37
38use crate::selfhosted::RunnerStatus;
39
40pub const ROOT: &str = "/var/lib/g1t-sandbox";
41/// The runner's folder (`/data` in its container): its registration, and
42/// the status file it keeps.
43pub const STATE_DIR: &str = "/var/lib/g1t-sandbox/state";
44pub const SETTINGS: &str = "/var/lib/g1t-sandbox/settings.json";
45pub const REGISTER_ENV: &str = "/var/lib/g1t-sandbox/register.env";
46pub const STATUS: &str = "/var/lib/g1t-sandbox/status.json";
47/// `drain` or `now`: asks the supervisor to stop.
48pub const STOP_FILE: &str = "/var/lib/g1t-sandbox/stop";
49/// A runner image the launcher put here for the supervisor to load
50/// (`sandbox up --runner-image-file`, for a build that is not released).
51pub const IMAGE_TAR: &str = "/var/lib/g1t-sandbox/runner-image.tar";
52pub const RUN_DIR: &str = "/run/g1t-sandbox";
53/// The VM's Docker Engine: not `/var/run/docker.sock`.
54pub const DOCKER_SOCKET: &str = "/run/g1t-sandbox/docker.sock";
55pub const PID_FILE: &str = "/run/g1t-sandbox/supervisor.pid";
56pub const RUNNER_CONTAINER: &str = "g1t-runner";
57/// How a status line starts on a VM's console.
58pub const STATUS_PREFIX: &str = "G1T-SANDBOX-STATUS ";
59
60/// The image's `/etc/wsl.conf`.
61pub const WSL_CONF: &str = include_str!("../../../../../deploy/sandbox/wsl.conf");
62
63/// What WSL mounts into every distribution that reaches Windows or other
64/// distributions, taken away at start.
65pub const WSL_CHANNELS: &[&str] = &["/mnt/wsl", "/mnt/wslg", "/tmp/.X11-unix", "/usr/lib/wsl/drivers", "/usr/lib/wsl/lib"];
66
67/// What the launcher tells the guest (`guest configure`, or the seed disk).
68#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
69pub struct Settings {
70 /// The site to register with, and a registration token: only until the
71 /// runner has registered.
72 #[serde(default)]
73 pub url: Option<String>,
74 #[serde(default)]
75 pub token: Option<String>,
76 #[serde(default)]
77 pub name: Option<String>,
78 #[serde(default)]
79 pub labels: Vec<String>,
80 #[serde(default)]
81 pub group: Option<String>,
82 /// The runner's container's share of the VM, and so its jobs'.
83 #[serde(default)]
84 pub cpus: u32,
85 #[serde(default)]
86 pub memory_mib: u64,
87 /// The runner's image, by digest from the signed release.
88 #[serde(default)]
89 pub runner_image: String,
90 #[serde(default)]
91 pub image_version: String,
92 /// For a VM, how to stop when it is shut down: `drain` or `now`.
93 #[serde(default)]
94 pub stop: Option<String>,
95}
96
97/// What `guest status` says, and the launcher reports.
98#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
99pub struct GuestStatus {
100 /// `starting`, `running`, `stopping`, `stopped` or `error`.
101 pub state: String,
102 #[serde(default)]
103 pub runner: Option<RunnerStatus>,
104 #[serde(default)]
105 pub error: Option<String>,
106 #[serde(default)]
107 pub image_version: Option<String>,
108 #[serde(default)]
109 pub updated_at: String,
110}
111
112/// A mount that is the computer's, or WSL's way to it, from
113/// `/proc/self/mountinfo`.
114pub fn host_mounts(mountinfo: &str) -> Vec<String> {
115 let mut found = Vec::new();
116 for line in mountinfo.lines() {
117 let fields: Vec<&str> = line.split_whitespace().collect();
118 if fields.len() < 5 {
119 continue;
120 }
121 let point = fields[4];
122 let fstype = line.split(" - ").nth(1).and_then(|rest| rest.split_whitespace().next()).unwrap_or("");
123 let shared_fs = matches!(fstype, "9p" | "drvfs" | "virtiofs" | "vboxsf" | "prl_fs" | "vmhgfs" | "fuse.vmhgfs-fuse" | "smb3" | "cifs");
124 let channel = WSL_CHANNELS.iter().any(|c| point == *c || point.starts_with(&format!("{c}/")));
125 if shared_fs || channel || point.starts_with("/mnt/") {
126 found.push(format!("{point} ({fstype})"));
127 }
128 }
129 found
130}
131
132/// What keeps the VM from being a sandbox: mounts of the computer's, an
133/// engine socket that is not this VM's own, WSL's interop with Windows.
134pub fn problems(mountinfo: &str, sockets: &[String], interop: bool, wsl_conf: Option<&str>) -> Vec<String> {
135 let mut problems: Vec<String> = host_mounts(mountinfo).into_iter().map(|m| format!("{m} is mounted from the computer, or is WSL's way to it")).collect();
136 for socket in sockets {
137 problems.push(format!("{socket} is a container engine's socket that is not this sandbox's own"));
138 }
139 if interop {
140 problems.push("WSL interop is on: Windows programs can be started from inside the sandbox".to_owned());
141 }
142 if let Some(conf) = wsl_conf
143 && conf.trim() != WSL_CONF.trim()
144 {
145 problems.push("/etc/wsl.conf is not the sandbox image's, which turns off Windows drives and programs".to_owned());
146 }
147 problems
148}
149
150/// The VM's default gateway, which is the computer, from `/proc/net/route`
151/// (addresses in little-endian hex).
152pub fn gateway(route: &str) -> Option<Ipv4Addr> {
153 route.lines().skip(1).find_map(|line| {
154 let fields: Vec<&str> = line.split_whitespace().collect();
155 if fields.len() > 2 && fields[1] == "00000000" {
156 u32::from_str_radix(fields[2], 16).ok().map(|n| Ipv4Addr::from(n.swap_bytes())).filter(|a| !a.is_unspecified())
157 } else {
158 None
159 }
160 })
161}
162
163/// The sandbox's `/etc/resolv.conf`: the resolvers WSL offered, else the
164/// gateway, which answers DNS on WSL's NAT.
165pub fn resolv_conf(offered: &str, gateway: Option<Ipv4Addr>) -> String {
166 let mut servers = nameservers(offered);
167 if servers.is_empty() {
168 servers.extend(gateway);
169 }
170 servers.iter().map(|s| format!("nameserver {s}\n")).collect()
171}
172
173pub fn nameservers(resolv: &str) -> Vec<Ipv4Addr> {
174 resolv
175 .lines()
176 .filter_map(|l| l.trim().strip_prefix("nameserver"))
177 .filter_map(|a| a.trim().parse().ok())
178 .collect()
179}
180
181/// `iptables` arguments, in order, for the `DOCKER-USER` chain, which
182/// every container's traffic passes: DNS to the VM's resolvers, and
183/// nothing else to the computer's address (the gateway), to a resolver
184/// the computer answers on, or to link-local addresses. The internet
185/// stays reachable, as a job needs.
186pub fn firewall_rules(gateway: Option<Ipv4Addr>, resolvers: &[Ipv4Addr]) -> Vec<Vec<String>> {
187 let s = |items: &[&str]| items.iter().map(|i| i.to_string()).collect::<Vec<String>>();
188 let mut rules = Vec::new();
189 let mut walled: Vec<Ipv4Addr> = gateway.into_iter().collect();
190 for resolver in resolvers {
191 if resolver.is_loopback() {
192 continue;
193 }
194 let address = resolver.to_string();
195 for proto in ["udp", "tcp"] {
196 rules.push(s(&["-d", &address, "-p", proto, "--dport", "53", "-j", "ACCEPT"]));
197 }
198 // A resolver that is the computer, or WSL's DNS tunnel, takes DNS
199 // and nothing else.
200 if (resolver.is_private() || resolver.is_link_local()) && !walled.contains(resolver) {
201 walled.push(*resolver);
202 }
203 }
204 for address in walled {
205 rules.push(s(&["-d", &address.to_string(), "-j", "DROP"]));
206 }
207 rules.push(s(&["-d", "169.254.0.0/16", "-j", "DROP"]));
208 rules
209}
210
211/// The VM's Docker Engine: Sysbox as a runtime, its own socket, address
212/// ranges away from Docker's defaults (the runner's own engine, inside,
213/// uses those).
214pub fn daemon_json() -> Value {
215 json!({
216 "hosts": [format!("unix://{DOCKER_SOCKET}")],
217 "pidfile": format!("{RUN_DIR}/dockerd.pid"),
218 "runtimes": { "sysbox-runc": { "path": "/usr/local/bin/sysbox-runc" } },
219 "bip": "172.29.254.1/24",
220 "default-address-pools": [{ "base": "172.28.0.0/16", "size": 24 }],
221 "log-driver": "json-file",
222 "log-opts": { "max-size": "20m", "max-file": "2" },
223 })
224}
225
226/// The registration's variables, for `register-and-run` in the runner's
227/// image: a file, so the token is on no command line.
228pub fn register_env(settings: &Settings) -> Result<String> {
229 let (Some(url), Some(token)) = (&settings.url, &settings.token) else { bail!("the sandbox's runner is not registered, and no --url and --token were given") };
230 let mut lines = vec![format!("G1T_RUNNER_URL={url}"), format!("G1T_RUNNER_TOKEN={token}")];
231 if let Some(name) = &settings.name {
232 lines.push(format!("G1T_RUNNER_NAME={name}"));
233 }
234 if !settings.labels.is_empty() {
235 lines.push(format!("G1T_RUNNER_LABELS={}", settings.labels.join(",")));
236 }
237 if let Some(group) = &settings.group {
238 lines.push(format!("G1T_RUNNER_GROUP={group}"));
239 }
240 if lines.iter().any(|l| l.contains('\n') || l.contains('\r')) {
241 bail!("a registration setting has a line break in it");
242 }
243 Ok(lines.join("\n") + "\n")
244}
245
246/// `docker run` for the runner in the VM.
247pub fn runner_run_args(settings: &Settings, register: bool) -> Vec<String> {
248 sysbox_run_args(settings, RUNNER_CONTAINER, STATE_DIR, register.then_some(REGISTER_ENV))
249}
250
251/// `docker run` for the runner: its image under Sysbox, `data` (a folder
252/// or a volume) as `/data`, no socket, no host network, its share of the
253/// machine, and no self-update (the launcher updates the whole sandbox).
254pub fn sysbox_run_args(settings: &Settings, name: &str, data: &str, env_file: Option<&str>) -> Vec<String> {
255 let mut args: Vec<String> = ["run", "-d", "--name", name, "--runtime=sysbox-runc", "--restart", "on-failure:5"]
256 .iter()
257 .map(|s| s.to_string())
258 .collect();
259 if settings.cpus > 0 {
260 args.extend(["--cpus".into(), settings.cpus.to_string()]);
261 }
262 if settings.memory_mib > 0 {
263 let memory = format!("{}m", settings.memory_mib);
264 args.extend(["--memory".into(), memory.clone(), "--memory-swap".into(), memory]);
265 }
266 args.extend(["-v".into(), format!("{data}:/data")]);
267 args.extend(["-e".into(), "G1T_RUNNER_NO_AUTO_UPDATE=1".into()]);
268 if let Some(file) = env_file {
269 args.extend(["--env-file".into(), file.into()]);
270 }
271 args.extend(["--label".into(), "sh.g1t.sandbox=runner".into()]);
272 args.push(settings.runner_image.clone());
273 args
274}
275
276/// Reads a seed disk: JSON, then zeros.
277pub fn read_seed(bytes: &[u8]) -> Result<Settings> {
278 let end = bytes.iter().position(|b| *b == 0).unwrap_or(bytes.len());
279 serde_json::from_slice(&bytes[..end]).context("the seed disk does not read")
280}
281
282/// A seed disk's bytes: the settings' JSON, padded to `size`.
283pub fn write_seed(settings: &Settings, size: usize) -> Result<Vec<u8>> {
284 let mut bytes = serde_json::to_vec(settings)?;
285 if bytes.len() >= size {
286 bail!("the sandbox's settings do not fit its seed disk");
287 }
288 bytes.resize(size, 0);
289 Ok(bytes)
290}
291
292/// The last status line on a VM's console.
293pub fn status_from_console(log: &str) -> Option<GuestStatus> {
294 log.lines().rev().find_map(|line| line.split_once(STATUS_PREFIX).and_then(|(_, json)| serde_json::from_str(json.trim()).ok()))
295}
296
297fn read_json<T: for<'a> Deserialize<'a>>(path: &str) -> Option<T> {
298 std::fs::read(path).ok().and_then(|bytes| serde_json::from_slice(&bytes).ok())
299}
300
301fn write_private(path: &str, bytes: &[u8]) -> Result<()> {
302 let temp = format!("{path}.new");
303 std::fs::write(&temp, bytes).with_context(|| format!("could not write {temp}"))?;
304 #[cfg(unix)]
305 {
306 use std::os::unix::fs::PermissionsExt;
307 std::fs::set_permissions(&temp, std::fs::Permissions::from_mode(0o600))?;
308 }
309 std::fs::rename(&temp, path).with_context(|| format!("could not write {path}"))?;
310 Ok(())
311}
312
313fn registered() -> bool {
314 Path::new(STATE_DIR).join("config.json").exists()
315}
316
317/// `g1t-runner sandbox guest <command>`.
318pub fn main(args: &[String]) -> Result<()> {
319 let command = args.first().map(String::as_str).unwrap_or("");
320 let value = |name: &str| args.windows(2).find(|w| w[0] == name).map(|w| w[1].clone());
321 match command {
322 "configure" => configure(&read_stdin()?),
323 "status" => {
324 println!("{}", serde_json::to_string(&current_status())?);
325 Ok(())
326 }
327 "export-state" => {
328 use std::io::Write;
329 let bytes = std::fs::read(Path::new(STATE_DIR).join("config.json")).context("the sandbox's runner is not registered")?;
330 std::io::stdout().write_all(&bytes)?;
331 Ok(())
332 }
333 "import-state" => {
334 std::fs::create_dir_all(STATE_DIR)?;
335 let bytes = read_stdin()?;
336 serde_json::from_slice::<Value>(&bytes).context("that is not a runner's configuration")?;
337 write_private(&format!("{STATE_DIR}/config.json"), &bytes)
338 }
339 "load-image" => {
340 std::fs::create_dir_all(ROOT)?;
341 let bytes = read_stdin()?;
342 write_private(IMAGE_TAR, &bytes)
343 }
344 "check" => {
345 let found = check();
346 if found.is_empty() {
347 println!("Nothing of the computer's reaches this sandbox.");
348 Ok(())
349 } else {
350 bail!("this sandbox can reach the computer:\n - {}", found.join("\n - "))
351 }
352 }
353 "stop" => stop(if args.iter().any(|a| a == "--now") { "now" } else { "drain" }, value("--wait").and_then(|w| w.parse().ok())),
354 "run" => run(value("--seed"), value("--state-disk"), value("--console")),
355 _ => bail!("guest takes configure, run, status, stop, check, export-state, import-state or load-image"),
356 }
357}
358
359fn read_stdin() -> Result<Vec<u8>> {
360 use std::io::Read;
361 let mut bytes = Vec::new();
362 std::io::stdin().read_to_end(&mut bytes)?;
363 // Windows' tools may put a byte-order mark before JSON.
364 if bytes.starts_with(&[0xEF, 0xBB, 0xBF]) {
365 bytes.drain(..3);
366 }
367 Ok(bytes)
368}
369
370/// Keeps what the launcher sent, over what was there: a token only until
371/// the runner has registered.
372fn configure(bytes: &[u8]) -> Result<()> {
373 let given: Settings = serde_json::from_slice(bytes).context("the sandbox's settings do not read")?;
374 std::fs::create_dir_all(STATE_DIR)?;
375 let mut settings: Settings = read_json(SETTINGS).unwrap_or_default();
376 if given.url.is_some() {
377 settings.url = given.url;
378 }
379 if given.token.is_some() && !registered() {
380 settings.token = given.token;
381 }
382 if given.name.is_some() {
383 settings.name = given.name;
384 }
385 if !given.labels.is_empty() {
386 settings.labels = given.labels;
387 }
388 if given.group.is_some() {
389 settings.group = given.group;
390 }
391 if given.cpus > 0 {
392 settings.cpus = given.cpus;
393 }
394 if given.memory_mib > 0 {
395 settings.memory_mib = given.memory_mib;
396 }
397 if !given.runner_image.is_empty() {
398 settings.runner_image = given.runner_image;
399 }
400 if !given.image_version.is_empty() {
401 settings.image_version = given.image_version;
402 }
403 write_private(SETTINGS, &serde_json::to_vec_pretty(&settings)?)
404}
405
406/// The supervisor's status, or `stopped` when it is not running.
407fn current_status() -> GuestStatus {
408 let status: Option<GuestStatus> = read_json(STATUS);
409 let alive = supervisor_alive();
410 match status {
411 Some(status) if alive || status.state == "error" || status.state == "stopped" => status,
412 Some(status) => GuestStatus { state: "stopped".into(), ..status },
413 None => GuestStatus { state: "stopped".into(), ..GuestStatus::default() },
414 }
415}
416
417fn supervisor_alive() -> bool {
418 std::fs::read_to_string(PID_FILE)
419 .ok()
420 .and_then(|p| p.trim().parse::<u32>().ok())
421 .is_some_and(|pid| Path::new(&format!("/proc/{pid}")).exists())
422}
423
424/// What reaches the computer from here, now.
425pub fn check() -> Vec<String> {
426 let mountinfo = std::fs::read_to_string("/proc/self/mountinfo").unwrap_or_default();
427 let mut sockets = Vec::new();
428 // `/var/run` is `/run`.
429 for dir in ["/mnt", "/run", "/tmp"] {
430 find_sockets(Path::new(dir), 3, &mut sockets);
431 }
432 let sockets: Vec<String> = sockets.into_iter().filter(|s| !own_socket(s)).collect();
433 let on_wsl = on_wsl();
434 // With interop off, WSL starts no interop server for the distribution's
435 // processes, and `WSL_INTEROP` names none.
436 let interop = on_wsl && std::env::var_os("WSL_INTEROP").is_some();
437 let conf = on_wsl.then(|| std::fs::read_to_string("/etc/wsl.conf").unwrap_or_default());
438 problems(&mountinfo, &sockets, interop, conf.as_deref())
439}
440
441/// The sandbox's own engine's sockets, and the containerd and Sysbox it
442/// runs.
443pub fn own_socket(path: &str) -> bool {
444 [RUN_DIR, "/run/docker/", "/run/containerd/", "/run/sysbox/"].iter().any(|own| path.starts_with(own))
445}
446
447#[cfg_attr(not(unix), allow(clippy::only_used_in_recursion))]
448fn find_sockets(dir: &Path, depth: u32, found: &mut Vec<String>) {
449 let Ok(entries) = std::fs::read_dir(dir) else { return };
450 for entry in entries.flatten() {
451 let Ok(kind) = entry.file_type() else { continue };
452 let path = entry.path();
453 #[cfg(unix)]
454 {
455 use std::os::unix::fs::FileTypeExt;
456 if kind.is_socket() {
457 let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
458 if ["docker.sock", "containerd.sock", "podman.sock", "docker.proxy.sock", "backend.sock", "filesystem.sock"].contains(&name) {
459 found.push(path.display().to_string());
460 }
461 continue;
462 }
463 }
464 if kind.is_dir() && depth > 0 {
465 find_sockets(&path, depth - 1, found);
466 }
467 }
468}
469
470fn on_wsl() -> bool {
471 std::env::var_os("WSL_DISTRO_NAME").is_some() || Path::new("/proc/sys/fs/binfmt_misc/WSLInterop").exists() || Path::new("/run/WSL").exists()
472}
473
474/// Asks the supervisor to stop (`drain`: once the runner's job is done),
475/// and waits up to `wait` seconds for it, or forever.
476fn stop(mode: &str, wait: Option<u64>) -> Result<()> {
477 if !supervisor_alive() {
478 return Ok(());
479 }
480 std::fs::write(STOP_FILE, mode)?;
481 let until = wait.map(|w| std::time::Instant::now() + std::time::Duration::from_secs(w));
482 while supervisor_alive() {
483 if until.is_some_and(|u| std::time::Instant::now() > u) {
484 bail!("the sandbox is still stopping");
485 }
486 std::thread::sleep(std::time::Duration::from_millis(500));
487 }
488 Ok(())
489}
490
491#[cfg(not(target_os = "linux"))]
492fn run(_seed: Option<String>, _state_disk: Option<String>, _console: Option<String>) -> Result<()> {
493 bail!("the sandbox's supervisor runs inside its Linux VM")
494}
495
496#[cfg(target_os = "linux")]
497fn run(seed: Option<String>, state_disk: Option<String>, console: Option<String>) -> Result<()> {
498 supervisor::run(seed, state_disk, console)
499}
500
501#[cfg(target_os = "linux")]
502mod supervisor {
503 use std::path::Path;
504 use std::process::{Child, Command, Stdio};
505 use std::sync::atomic::Ordering;
506 use std::time::{Duration, Instant};
507
508 use anyhow::{Context, Result, bail};
509
510 use super::*;
511 use crate::selfhosted::{STOP, g1t_now};
512
513 extern "C" fn on_term(_: libc::c_int) {
514 STOP.store(true, Ordering::SeqCst);
515 }
516
517 fn now() -> String {
518 g1t_now(std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0))
519 }
520
521 /// A clean environment: WSL's session points `XDG_RUNTIME_DIR` and
522 /// the display variables into /mnt/wslg, which is gone.
523 fn clean(command: &mut Command) -> &mut Command {
524 command.env_clear().env("PATH", "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin").env("HOME", "/root")
525 }
526
527 fn docker() -> Command {
528 let mut command = Command::new("docker");
529 clean(&mut command).env("DOCKER_HOST", format!("unix://{DOCKER_SOCKET}"));
530 command
531 }
532
533 fn ok(command: &mut Command) -> bool {
534 command.stdout(Stdio::null()).stderr(Stdio::null()).status().is_ok_and(|s| s.success())
535 }
536
537 struct Say {
538 console: Option<String>,
539 version: Option<String>,
540 }
541
542 impl Say {
543 fn status(&self, state: &str, runner: Option<RunnerStatus>, error: Option<String>) {
544 let status = GuestStatus { state: state.into(), runner, error, image_version: self.version.clone(), updated_at: now() };
545 let Ok(bytes) = serde_json::to_vec(&status) else { return };
546 let _ = write_private(STATUS, &bytes);
547 if let Some(console) = &self.console {
548 let _ = std::fs::OpenOptions::new()
549 .append(true)
550 .open(console)
551 .and_then(|mut c| std::io::Write::write_all(&mut c, format!("{STATUS_PREFIX}{}\n", String::from_utf8_lossy(&bytes)).as_bytes()));
552 }
553 }
554 }
555
556 fn daemon(program: &str, args: &[&str], log: &str) -> Result<Child> {
557 let out = std::fs::File::create(format!("{RUN_DIR}/{log}"))?;
558 let mut command = Command::new(program);
559 clean(&mut command)
560 .args(args)
561 .stdout(out.try_clone()?)
562 .stderr(out)
563 .stdin(Stdio::null())
564 .spawn()
565 .with_context(|| format!("could not start {program}"))
566 }
567
568 fn wait_for(what: &str, limit: Duration, mut ready: impl FnMut() -> bool) -> Result<()> {
569 let until = Instant::now() + limit;
570 while !ready() {
571 if Instant::now() > until {
572 bail!("{what} did not start");
573 }
574 std::thread::sleep(Duration::from_millis(250));
575 }
576 Ok(())
577 }
578
579 /// WSL's ways to Windows and to other distributions, taken away.
580 fn wsl_prepare() -> Result<()> {
581 // The resolvers WSL offers (its DNS tunnel, or the gateway), in a
582 // file of the sandbox's own: WSL's is in /mnt/wsl, unmounted next.
583 let offered = ["/mnt/wsl/resolv.conf", "/etc/resolv.conf"].iter().find_map(|p| std::fs::read_to_string(p).ok()).unwrap_or_default();
584 let route = std::fs::read_to_string("/proc/net/route").unwrap_or_default();
585 let _ = std::fs::remove_file("/etc/resolv.conf");
586 std::fs::write("/etc/resolv.conf", resolv_conf(&offered, gateway(&route)))?;
587 for point in WSL_CHANNELS {
588 let _ = Command::new("umount").args(["-R", "-l", point]).stdout(Stdio::null()).stderr(Stdio::null()).status();
589 }
590 // The GPU's paravirtual device: a way to Windows' driver.
591 let _ = std::fs::remove_file("/dev/dxg");
592 Ok(())
593 }
594
595 fn mount_state_disk(device: &str) -> Result<()> {
596 std::fs::create_dir_all(STATE_DIR)?;
597 let blank = !Command::new("blkid").arg(device).output().is_ok_and(|o| o.status.success() && !o.stdout.is_empty());
598 if blank && !ok(Command::new("mkfs.ext4").args(["-q", "-L", "g1t-state", device])) {
599 bail!("could not format the state disk {device}");
600 }
601 if !ok(Command::new("mount").args([device, STATE_DIR])) {
602 bail!("could not mount the state disk {device}");
603 }
604 Ok(())
605 }
606
607 fn read_seed_device(device: &str) -> Result<Settings> {
608 // The launcher may have rewritten it since boot: read past the cache.
609 let _ = Command::new("blockdev").args(["--flushbufs", device]).status();
610 let bytes = std::fs::read(device).with_context(|| format!("could not read the seed disk {device}"))?;
611 read_seed(&bytes)
612 }
613
614 fn firewall() -> Result<()> {
615 let route = std::fs::read_to_string("/proc/net/route").unwrap_or_default();
616 let resolv = std::fs::read_to_string("/etc/resolv.conf").unwrap_or_default();
617 let rules = firewall_rules(gateway(&route), &nameservers(&resolv));
618 let _ = Command::new("iptables").args(["-N", "DOCKER-USER"]).stderr(Stdio::null()).status();
619 // Only these rules, however many times the sandbox has started.
620 let _ = Command::new("iptables").args(["-F", "DOCKER-USER"]).status();
621 // Inserted last first, so they end up in order at the chain's top.
622 for (at, rule) in rules.iter().enumerate() {
623 let position = (at + 1).to_string();
624 let mut args = vec!["-I", "DOCKER-USER", position.as_str()];
625 args.extend(rule.iter().map(String::as_str));
626 if !ok(Command::new("iptables").args(&args)) {
627 bail!("could not add the rule that keeps containers off the computer: iptables {}", args.join(" "));
628 }
629 }
630 Ok(())
631 }
632
633 fn container_state() -> Option<(bool, i32)> {
634 let out = docker().args(["inspect", "-f", "{{.State.Running}} {{.State.ExitCode}}", RUNNER_CONTAINER]).output().ok()?;
635 if !out.status.success() {
636 return None;
637 }
638 let text = String::from_utf8_lossy(&out.stdout);
639 let mut parts = text.split_whitespace();
640 Some((parts.next() == Some("true"), parts.next().and_then(|c| c.parse().ok()).unwrap_or(0)))
641 }
642
643 /// The runner's last words: its last line of output.
644 fn logs_tail() -> String {
645 docker()
646 .args(["logs", "--tail", "1", RUNNER_CONTAINER])
647 .output()
648 .map(|o| format!("{}{}", String::from_utf8_lossy(&o.stdout), String::from_utf8_lossy(&o.stderr)).trim().to_owned())
649 .unwrap_or_default()
650 .trim_start_matches("g1t-runner: ")
651 .to_owned()
652 }
653
654 fn start_runner(settings: &Settings) -> Result<()> {
655 if let Some((true, _)) = container_state() {
656 return Ok(());
657 }
658 let _ = docker().args(["rm", "-f", RUNNER_CONTAINER]).stdout(Stdio::null()).stderr(Stdio::null()).status();
659 // What an earlier runner said is not what this one says.
660 let _ = std::fs::remove_file(format!("{STATE_DIR}/status.json"));
661 let register = !registered();
662 if register {
663 write_private(REGISTER_ENV, register_env(settings)?.as_bytes())?;
664 }
665 let out = docker().args(runner_run_args(settings, register)).output().context("could not run docker")?;
666 if !out.status.success() {
667 bail!("could not start the runner: {}", String::from_utf8_lossy(&out.stderr).trim());
668 }
669 Ok(())
670 }
671
672 pub fn run(seed: Option<String>, state_disk: Option<String>, console: Option<String>) -> Result<()> {
673 unsafe {
674 libc::signal(libc::SIGTERM, on_term as *const () as libc::sighandler_t);
675 libc::signal(libc::SIGINT, on_term as *const () as libc::sighandler_t);
676 }
677 std::fs::create_dir_all(RUN_DIR)?;
678 std::fs::create_dir_all(ROOT)?;
679 std::fs::write(PID_FILE, std::process::id().to_string())?;
680 let _ = std::fs::remove_file(STOP_FILE);
681 let version = std::fs::read_to_string("/etc/g1t-sandbox-version").ok().map(|v| v.trim().to_owned());
682 let say = Say { console, version };
683 say.status("starting", None, None);
684 let outcome = supervise(&say, seed, state_disk);
685 if let Err(error) = &outcome {
686 say.status("error", None, Some(format!("{error:#}")));
687 }
688 let _ = std::fs::remove_file(PID_FILE);
689 outcome
690 }
691
692 fn supervise(say: &Say, seed: Option<String>, state_disk: Option<String>) -> Result<()> {
693 if on_wsl() {
694 wsl_prepare()?;
695 }
696 if let Some(device) = &state_disk {
697 // A VM's system disk was grown to the size asked for: its file
698 // system grows to fill it.
699 let _ = Command::new("resize2fs").arg("/dev/vda").stdout(Stdio::null()).stderr(Stdio::null()).status();
700 mount_state_disk(device)?;
701 }
702 std::fs::create_dir_all(STATE_DIR)?;
703 if let Some(device) = &seed {
704 let settings = read_seed_device(device)?;
705 configure(&serde_json::to_vec(&settings)?)?;
706 }
707 let found = check();
708 if !found.is_empty() {
709 bail!("this sandbox can reach the computer, so it will not run jobs:\n - {}", found.join("\n - "));
710 }
711 let settings: Settings = read_json(SETTINGS).context("the sandbox has no settings: run `g1t-runner sandbox up` on the computer")?;
712 if settings.runner_image.is_empty() {
713 bail!("the sandbox's settings name no runner image");
714 }
715
716 let mut daemons: Vec<Child> = Vec::new();
717 let _ = std::fs::remove_file("/run/sysbox/sysmgr.sock");
718 daemons.push(daemon("sysbox-mgr", &[], "sysbox-mgr.log")?);
719 wait_for("Sysbox's manager", Duration::from_secs(30), || Path::new("/run/sysbox/sysmgr.sock").exists())?;
720 daemons.push(daemon("sysbox-fs", &[], "sysbox-fs.log")?);
721 wait_for("Sysbox's file system", Duration::from_secs(30), || Path::new("/run/sysbox/sysfs.sock").exists())?;
722 std::fs::write("/etc/docker/daemon.json", serde_json::to_vec_pretty(&daemon_json())?)?;
723 daemons.push(daemon("dockerd", &[], "dockerd.log")?);
724 let outcome = (|| -> Result<()> {
725 wait_for("The sandbox's Docker Engine", Duration::from_secs(60), || ok(docker().args(["version"])))?;
726 firewall()?;
727 if Path::new(IMAGE_TAR).exists() {
728 let loaded = docker().args(["load", "-i", IMAGE_TAR]).stdout(Stdio::null()).status();
729 let _ = std::fs::remove_file(IMAGE_TAR);
730 if !loaded.is_ok_and(|s| s.success()) {
731 bail!("could not load the runner image the launcher gave");
732 }
733 }
734 if !ok(docker().args(["image", "inspect", &settings.runner_image])) && !ok(docker().args(["pull", "-q", &settings.runner_image])) {
735 bail!("could not fetch the runner's image {}", settings.runner_image);
736 }
737 start_runner(&settings)?;
738 watch(say)
739 })();
740 // Everything stops with the supervisor; the runner first, so it can
741 // say how its job ended.
742 let _ = docker().args(["stop", "-t", "60", RUNNER_CONTAINER]).stdout(Stdio::null()).stderr(Stdio::null()).status();
743 for child in daemons.iter_mut().rev() {
744 unsafe {
745 libc::kill(child.id() as libc::pid_t, libc::SIGTERM);
746 }
747 let until = Instant::now() + Duration::from_secs(20);
748 while matches!(child.try_wait(), Ok(None)) && Instant::now() < until {
749 std::thread::sleep(Duration::from_millis(200));
750 }
751 let _ = child.kill();
752 let _ = child.wait();
753 }
754 if outcome.is_ok() {
755 say.status("stopped", None, None);
756 }
757 outcome
758 }
759
760 /// Says how the runner is until asked to stop, then stops it.
761 fn watch(say: &Say) -> Result<()> {
762 let data = Path::new(STATE_DIR);
763 let mut asked: Option<String> = None;
764 let mut tried_again = Instant::now();
765 loop {
766 if asked.is_none() {
767 let mode = std::fs::read_to_string(STOP_FILE).ok().map(|m| m.trim().to_owned());
768 asked = mode.or_else(|| STOP.load(Ordering::SeqCst).then(|| read_json::<Settings>(SETTINGS).and_then(|s| s.stop).unwrap_or_else(|| "now".into())));
769 if let Some(mode) = &asked {
770 // The runner reads these in its folder (selfhosted/mod.rs).
771 std::fs::write(data.join(if mode == "drain" { "drain" } else { "stop" }), "")?;
772 }
773 }
774 if registered() && Path::new(REGISTER_ENV).exists() {
775 // Registered: the token is no longer kept.
776 let _ = std::fs::remove_file(REGISTER_ENV);
777 if let Some(mut settings) = read_json::<Settings>(SETTINGS) {
778 settings.token = None;
779 let _ = write_private(SETTINGS, &serde_json::to_vec_pretty(&settings)?);
780 }
781 }
782 let runner: Option<RunnerStatus> = read_json(&format!("{STATE_DIR}/status.json"));
783 match container_state() {
784 Some((true, _)) => {
785 let state = if asked.is_some() { "stopping" } else { "running" };
786 say.status(state, runner, None);
787 }
788 Some((false, code)) if asked.is_some() || code == 0 => {
789 let _ = std::fs::remove_file(data.join("stop"));
790 let _ = std::fs::remove_file(data.join("drain"));
791 return Ok(());
792 }
793 Some((false, code)) => {
794 say.status("error", runner, Some(format!("The runner stopped ({code}): {}", logs_tail())));
795 if tried_again.elapsed() > Duration::from_secs(30) {
796 tried_again = Instant::now();
797 let _ = docker().args(["start", RUNNER_CONTAINER]).stdout(Stdio::null()).stderr(Stdio::null()).status();
798 }
799 }
800 None => bail!("the runner's container is gone"),
801 }
802 std::thread::sleep(Duration::from_secs(2));
803 }
804 }
805}
806
807#[cfg(test)]
808mod tests {
809 use super::*;
810
811 const WSL_MOUNTINFO: &str = "\
812353 361 0:29 / /usr/lib/modules/6.6.87.2-microsoft-standard-WSL2 rw,nosuid,nodev,noatime - overlay none rw,lowerdir=/modules
813354 361 0:32 / /mnt/wsl rw,relatime shared:1 - tmpfs none rw
814355 354 0:61 /guest-services /mnt/wsl/docker-desktop/shared-sockets/guest-services rw,nosuid,nodev shared:18 - tmpfs none rw,mode=755
815358 361 0:34 / /usr/lib/wsl/drivers ro,nosuid,nodev,noatime - 9p drivers ro,aname=drivers;fmask=222;dmask=222
816361 343 8:96 / / rw,relatime - ext4 /dev/sdg rw,discard,errors=remount-ro,data=ordered
817362 361 0:87 / /mnt/wslg rw,relatime shared:32 - tmpfs none rw
818389 361 0:104 / /usr/lib/wsl/lib rw,nosuid,nodev,noatime - overlay none rw,lowerdir=/gpu_lib_packaged:/gpu_lib_inbox
819393 361 0:108 / /proc rw,nosuid,nodev,noexec,noatime shared:46 - proc proc rw
820411 361 0:87 /.X11-unix /tmp/.X11-unix ro,relatime shared:32 - tmpfs none rw
821";
822
823 const CLEAN: &str = "\
824361 343 8:96 / / rw,relatime - ext4 /dev/sdg rw,discard,errors=remount-ro,data=ordered
825393 361 0:108 / /proc rw,nosuid,nodev,noexec,noatime shared:46 - proc proc rw
826395 361 0:110 / /run rw,nosuid,nodev shared:48 - tmpfs none rw,mode=755
827";
828
829 #[test]
830 fn what_wsl_mounts_is_found_and_a_cleaned_distribution_is_clean() {
831 let found = host_mounts(WSL_MOUNTINFO);
832 for point in ["/mnt/wsl ", "/mnt/wsl/docker-desktop", "/usr/lib/wsl/drivers", "/mnt/wslg", "/usr/lib/wsl/lib", "/tmp/.X11-unix"] {
833 assert!(found.iter().any(|f| f.starts_with(point)), "{point} in {found:?}");
834 }
835 assert!(!found.iter().any(|f| f.starts_with("/usr/lib/modules")));
836 assert!(host_mounts(CLEAN).is_empty());
837 let drive = "500 361 0:90 / /mnt/c rw,noatime - 9p C:\\ rw,aname=drvfs;path=C:\\";
838 assert_eq!(host_mounts(drive), vec!["/mnt/c (9p)".to_owned()]);
839 assert_eq!(host_mounts("501 361 0:91 / /work rw - virtiofs share rw").len(), 1);
840 }
841
842 #[test]
843 fn the_sandbox_refuses_what_reaches_the_computer() {
844 assert!(problems(CLEAN, &[], false, Some(WSL_CONF)).is_empty());
845 assert!(problems(CLEAN, &[], false, None).is_empty());
846 let all = problems(WSL_MOUNTINFO, &["/mnt/wsl/docker-desktop/shared-sockets/host-services/docker.proxy.sock".into()], true, Some("[automount]\nenabled=true\n"));
847 assert!(all.iter().any(|p| p.contains("docker.proxy.sock")));
848 assert!(all.iter().any(|p| p.contains("interop is on")));
849 assert!(all.iter().any(|p| p.contains("/etc/wsl.conf")));
850 }
851
852 #[test]
853 fn the_image_turns_off_every_way_to_windows() {
854 let conf = WSL_CONF.replace(' ', "");
855 for setting in ["[automount]\nenabled=false", "mountFsTab=false", "[interop]\nenabled=false", "appendWindowsPath=false", "default=root"] {
856 assert!(conf.contains(setting), "{setting}");
857 }
858 }
859
860 #[test]
861 fn the_gateway_and_resolvers_read() {
862 let route = "Iface\tDestination\tGateway \tFlags\tRefCnt\tUse\tMetric\tMask\t\tMTU\tWindow\tIRTT\n\
863eth0\t00000000\t01E01EAC\t0003\t0\t0\t0\t00000000\t0\t0\t0\n\
864eth0\t00E01EAC\t00000000\t0001\t0\t0\t0\t00F0FFFF\t0\t0\t0\n";
865 assert_eq!(gateway(route), Some(Ipv4Addr::new(172, 30, 224, 1)));
866 assert_eq!(gateway("Iface\tDestination\n"), None);
867 assert_eq!(nameservers("# x\nnameserver 10.255.255.254\nsearch lan\nnameserver ::1\n"), vec![Ipv4Addr::new(10, 255, 255, 254)]);
868 assert_eq!(resolv_conf("nameserver 10.255.255.254\n", None), "nameserver 10.255.255.254\n");
869 assert_eq!(resolv_conf("", Some(Ipv4Addr::new(172, 30, 224, 1))), "nameserver 172.30.224.1\n");
870 }
871
872 #[test]
873 fn containers_get_dns_and_nothing_else_of_the_computer() {
874 let gw = Ipv4Addr::new(172, 30, 224, 1);
875 let rules = firewall_rules(Some(gw), &[Ipv4Addr::new(10, 255, 255, 254), Ipv4Addr::new(1, 1, 1, 1), Ipv4Addr::LOCALHOST]);
876 let text: Vec<String> = rules.iter().map(|r| r.join(" ")).collect();
877 assert_eq!(text[0], "-d 10.255.255.254 -p udp --dport 53 -j ACCEPT");
878 assert!(text.contains(&"-d 1.1.1.1 -p tcp --dport 53 -j ACCEPT".to_owned()));
879 let first_drop = text.iter().position(|r| r.ends_with("DROP")).unwrap();
880 assert!(text[..first_drop].iter().all(|r| r.ends_with("ACCEPT")), "accepts come first: {text:?}");
881 assert!(text.contains(&"-d 172.30.224.1 -j DROP".to_owned()));
882 assert!(text.contains(&"-d 10.255.255.254 -j DROP".to_owned()));
883 assert!(!text.contains(&"-d 1.1.1.1 -j DROP".to_owned()), "a public resolver is the internet");
884 assert!(!text.iter().any(|r| r.contains("127.0.0.1")));
885 assert_eq!(text.last().unwrap(), "-d 169.254.0.0/16 -j DROP");
886 }
887
888 #[test]
889 fn the_runner_runs_under_sysbox_with_no_socket_and_no_self_update() {
890 let settings = Settings { cpus: 4, memory_mib: 8192, runner_image: "g1t.sh/flagon-io/g1t-runner@sha256:ab".into(), ..Settings::default() };
891 let args = runner_run_args(&settings, false).join(" ");
892 assert!(args.starts_with("run -d --name g1t-runner --runtime=sysbox-runc"));
893 assert!(args.contains("--cpus 4 --memory 8192m --memory-swap 8192m"));
894 assert!(args.contains("-v /var/lib/g1t-sandbox/state:/data"));
895 assert!(args.contains("G1T_RUNNER_NO_AUTO_UPDATE=1"));
896 assert!(!args.contains("docker.sock") && !args.contains("--privileged") && !args.contains("--network host"));
897 assert!(!args.contains("--env-file"));
898 assert!(args.ends_with("g1t.sh/flagon-io/g1t-runner@sha256:ab"));
899 assert!(runner_run_args(&settings, true).join(" ").contains("--env-file /var/lib/g1t-sandbox/register.env"));
900 }
901
902 #[test]
903 fn registration_goes_by_file_and_refuses_line_breaks() {
904 let mut settings = Settings {
905 url: Some("https://g1t.sh".into()),
906 token: Some("g1trt_x".into()),
907 name: Some("box-linux".into()),
908 labels: vec!["gpu".into()],
909 ..Settings::default()
910 };
911 let env = register_env(&settings).unwrap();
912 assert_eq!(env, "G1T_RUNNER_URL=https://g1t.sh\nG1T_RUNNER_TOKEN=g1trt_x\nG1T_RUNNER_NAME=box-linux\nG1T_RUNNER_LABELS=gpu\n");
913 settings.name = Some("a\nG1T_RUNNER_URL=https://evil".into());
914 assert!(register_env(&settings).is_err());
915 assert!(register_env(&Settings::default()).is_err());
916 }
917
918 #[test]
919 fn a_seed_disk_and_a_console_line_round_trip() {
920 let settings = Settings { runner_image: "img".into(), cpus: 2, stop: Some("drain".into()), ..Settings::default() };
921 let bytes = write_seed(&settings, 4096).unwrap();
922 assert_eq!(bytes.len(), 4096);
923 assert_eq!(read_seed(&bytes).unwrap(), settings);
924 assert!(write_seed(&settings, 8).is_err());
925 let log = format!("[ 1.0] booting\n{STATUS_PREFIX}{{\"state\":\"starting\"}}\nnoise\n{STATUS_PREFIX}{{\"state\":\"running\",\"updated_at\":\"t\"}}\n");
926 assert_eq!(status_from_console(&log).unwrap().state, "running");
927 assert!(status_from_console("nothing").is_none());
928 }
929
930 #[test]
931 fn the_vm_engine_has_sysbox_and_its_own_socket() {
932 let config = daemon_json();
933 assert_eq!(config["hosts"][0], "unix:///run/g1t-sandbox/docker.sock");
934 assert_eq!(config["runtimes"]["sysbox-runc"]["path"], "/usr/local/bin/sysbox-runc");
935 }
936}