introduce rust

src/cmd/init.rs

@@ -0,0 +1,271 @@
+use eyre::{format_err, Result};
+use log::{error, info, warn};
+use std::collections::BTreeSet as Set;
+use std::os::unix::fs::symlink;
+use tokio::sync::Mutex;
+use tokio::{fs, process::Command};
+
+use crate::{bootstrap::config::Config, cmd::version::version_string, dklog, input};
+
+mod bootstrap;
+mod dmcrypt;
+mod lvm;
+mod networks;
+mod sshd;
+
+// devices used by any block target (DM, LVM...)
+static USED_DEVS: Mutex<Set<String>> = Mutex::const_new(Set::new());
+
+pub async fn run() {
+    if std::process::id() != 1 {
+        error!("init must run as PID 1, not {}", std::process::id());
+        std::process::exit(1);
+    }
+
+    unsafe {
+        use std::env;
+        env::set_var("PATH", "/bin:/sbin:/usr/bin:/usr/sbin");
+        env::set_var("HOME", "/root");
+    }
+
+    info!("Welcome to {}", version_string());
+
+    let uname = nix::sys::utsname::uname().expect("uname should work");
+    let kernel_version = uname.release().to_string_lossy();
+    info!("Linux version {kernel_version}");
+
+    let cfg: Config = retry(async || {
+        let cfg = (fs::read("config.yaml").await)
+            .map_err(|e| format_err!("failed to read config: {e}"))?;
+        serde_yaml::from_slice(cfg.as_slice())
+            .map_err(|e| format_err!("failed to parse config: {e}"))
+    })
+    .await;
+
+    info!("config loaded");
+    info!("anti-phishing-code: {}", cfg.anti_phishing_code);
+
+    // seem to occasionaly race with Child::wait
+    // tokio::spawn(child_reaper());
+
+    // open input channels
+    tokio::spawn(crate::input::answer_requests_from_stdin());
+    tokio::spawn(crate::input::answer_requests_from_socket());
+
+    // mount basic filesystems
+    mount("none", "/proc", "proc", None).await;
+    mount("none", "/sys", "sysfs", None).await;
+    mount("none", "/dev", "devtmpfs", None).await;
+    mount("none", "/dev/pts", "devpts", Some("gid=5,mode=620")).await;
+
+    // mount modules
+    if let Some(ref modules) = cfg.modules {
+        retry_or_ignore(async || {
+            info!("mounting modules");
+            mount(modules, "/modules", "squashfs", None).await;
+
+            fs::create_dir_all("/lib/modules").await?;
+            let modules_path = &format!("/modules/lib/modules/{kernel_version}");
+
+            if !std::fs::exists(modules_path)? {
+                return Err(format_err!(
+                    "invalid modules package: {modules_path} should exist"
+                ));
+            }
+
+            symlink(modules_path, format!("/lib/modules/{kernel_version}"))?;
+            Ok(())
+        })
+        .await;
+    } else {
+        warn!("modules NOT mounted (not configured)");
+    }
+
+    // init devices
+    info!("initializing devices");
+    start_daemon("udevd", &[]).await;
+
+    exec("udevadm", &["trigger", "-c", "add", "-t", "devices"]).await;
+    exec("udevadm", &["trigger", "-c", "add", "-t", "subsystems"]).await;
+    exec("udevadm", &["settle"]).await;
+
+    // Wireguard VPN
+    // TODO startVPN()
+
+    // SSH service
+    sshd::start(&cfg).await;
+
+    // networks
+    networks::setup(&cfg).await;
+
+    // pre-lvm dmcrypt devs
+    dmcrypt::setup(&cfg.pre_lvm_crypt).await;
+
+    // LVM
+    lvm::setup(&cfg).await;
+
+    // post-lvm dmcrypt devs
+    dmcrypt::setup(&cfg.crypt).await;
+
+    // bootstrap the system
+    bootstrap::bootstrap(cfg).await;
+
+    // finalize
+    retry(async || switch_root("/system").await).await;
+}
+
+async fn switch_root(root: &str) -> Result<()> {
+    use std::ffi::CString;
+
+    macro_rules! c {
+        ($s:expr) => {
+            CString::new($s)?.as_c_str()
+        };
+    }
+
+    info!("switching root");
+    dklog::LOG.close().await;
+
+    nix::unistd::execv(
+        c!("/sbin/switch_root"),
+        &[
+            c!("switch_root"),
+            c!("-c"),
+            c!("/dev/console"),
+            c!(root),
+            c!("/sbin/init"),
+        ],
+    )?;
+    Ok(())
+}
+
+async fn mount(src: &str, dst: &str, fstype: &str, opts: Option<&str>) {
+    if let Err(e) = fs::create_dir_all(dst).await {
+        error!("failed to create dir {dst}: {e}");
+    }
+
+    match fstype {
+        "ext4" => {
+            exec("fsck", &["-p", src]).await;
+        }
+        _ => {}
+    }
+
+    let mut args = vec![src, dst];
+    if !fstype.is_empty() {
+        args.extend(&["-t", fstype]);
+    }
+    if let Some(opts) = opts {
+        args.extend(["-o", opts]);
+    }
+    exec("mount", &args).await;
+}
+
+async fn start_daemon(prog: &str, args: &[&str]) {
+    let (cmd_str, mut cmd) = cmd_str(prog, args);
+    retry_or_ignore(async || {
+        info!("starting as daemon: {cmd_str}");
+        cmd.spawn()?;
+        Ok(())
+    })
+    .await;
+}
+
+async fn try_exec(prog: &str, args: &[&str]) -> Result<()> {
+    let (cmd_str, mut cmd) = cmd_str(prog, args);
+    info!("# {cmd_str}");
+
+    let s = cmd.status().await?;
+    if s.success() {
+        Ok(())
+    } else {
+        Err(format_err!("command failed: {s}"))
+    }
+}
+
+async fn exec(prog: &str, args: &[&str]) {
+    retry_or_ignore(async || try_exec(prog, args).await).await;
+}
+
+async fn retry_or_ignore(mut action: impl AsyncFnMut() -> Result<()>) {
+    loop {
+        match action().await {
+            Ok(_) => return,
+            Err(e) => {
+                error!("{e}");
+
+                match input::read_choice(["[r]etry", "[i]gnore", "[s]hell"]).await {
+                    'r' => {}
+                    'i' => return,
+                    's' => exec_shell().await,
+                    _ => unreachable!(),
+                }
+            }
+        }
+    }
+}
+
+async fn retry<T>(mut action: impl AsyncFnMut() -> Result<T>) -> T {
+    loop {
+        match action().await {
+            Ok(v) => return v,
+            Err(e) => {
+                error!("{e}");
+
+                match input::read_choice(["[r]etry", "[s]hell"]).await {
+                    'r' => {}
+                    's' => exec_shell().await,
+                    _ => unreachable!(),
+                }
+            }
+        }
+    }
+}
+
+async fn exec_shell() {
+    let mut child = match Command::new("ash").spawn() {
+        Ok(c) => c,
+        Err(e) => {
+            error!("failed to exec shell: {e}");
+            return;
+        }
+    };
+
+    let _ = child.wait().await;
+}
+
+fn cmd_str(prog: &str, args: &[&str]) -> (String, Command) {
+    use std::borrow::Cow;
+
+    let mut buf = String::new();
+
+    buf.push_str(&shell_escape::escape(Cow::Borrowed(prog)));
+
+    for &arg in args {
+        buf.push(' ');
+        buf.push_str(&shell_escape::escape(Cow::Borrowed(arg)));
+    }
+
+    let mut cmd = Command::new(prog);
+    cmd.args(args);
+
+    (buf, cmd)
+}
+
+#[allow(unused)]
+async fn child_reaper() {
+    use nix::sys::wait::{waitpid, WaitPidFlag};
+    use nix::unistd::Pid;
+    use tokio::signal::unix::{signal, SignalKind};
+
+    let Ok(mut sigs) =
+        signal(SignalKind::child()).inspect_err(|e| warn!("failed to setup SIGCHLD handler: {e}"))
+    else {
+        return;
+    };
+
+    loop {
+        sigs.recv().await;
+        let _ = waitpid(Some(Pid::from_raw(-1)), Some(WaitPidFlag::WNOHANG));
+    }
+}

src/cmd/init/bootstrap.rs

@@ -0,0 +1,310 @@
+use eyre::{format_err, Result};
+use log::{info, warn};
+use std::path::Path;
+use tokio::{
+    fs,
+    io::{AsyncBufReadExt, BufReader},
+};
+
+use super::{exec, mount, retry, retry_or_ignore, try_exec};
+use crate::bootstrap::config::Config;
+use crate::dkl;
+
+pub async fn bootstrap(cfg: Config) {
+    let bs = cfg.bootstrap;
+
+    retry_or_ignore(async || {
+        fs::create_dir_all("/boostrap").await?;
+        mount(&bs.dev, "/bootstrap", "auto", None).await;
+        Ok(())
+    })
+    .await;
+
+    let boot_version = "current"; // FIXME need to parse kernel cmdline
+    let base_dir = &format!("/bootstrap/{boot_version}");
+
+    retry_or_ignore(async || {
+        if !fs::try_exists(&base_dir).await? {
+            info!("creating {base_dir}");
+            fs::create_dir_all(&base_dir).await?
+        }
+        Ok(())
+    })
+    .await;
+
+    let sys_cfg: dkl::Config = retry(async || {
+        let sys_cfg_bytes = seed_config(base_dir, &bs.seed).await?;
+        Ok(serde_yaml::from_slice(&sys_cfg_bytes)?)
+    })
+    .await;
+
+    mount_system(&sys_cfg, base_dir).await;
+
+    retry_or_ignore(async || apply_files(&sys_cfg.files, "/system").await).await;
+
+    apply_groups(&sys_cfg.groups, "/system").await;
+    apply_users(&sys_cfg.users, "/system").await;
+
+    // TODO VPNs
+
+    mount_filesystems(&sys_cfg.mounts, "/system").await;
+
+    retry_or_ignore(async || {
+        info!("setting up root user");
+        setup_root_user(&sys_cfg.root_user, "/system").await
+    })
+    .await;
+}
+
+async fn seed_config(base_dir: &str, seed_url: &Option<String>) -> Result<Vec<u8>> {
+    let cfg_path = &format!("{base_dir}/config.yaml");
+
+    if fs::try_exists(cfg_path).await? {
+        return Ok(fs::read(cfg_path).await?);
+    }
+
+    let bs_tar = "/bootstrap.tar";
+    if !fs::try_exists(bs_tar).await? {
+        if let Some(seed_url) = seed_url.as_ref() {
+            fetch_bootstrap(seed_url, bs_tar).await?;
+        } else {
+            return Err(format_err!(
+                "no {cfg_path}, no {bs_tar} and no seed, can't bootstrap"
+            ));
+        }
+    }
+
+    try_exec("tar", &["xf", bs_tar, "-C", base_dir]).await?;
+
+    if !fs::try_exists(cfg_path).await? {
+        return Err(format_err!("{cfg_path} does not exist after seeding"));
+    }
+
+    Ok(fs::read(cfg_path).await?)
+}
+
+async fn fetch_bootstrap(seed_url: &str, output_file: &str) -> Result<()> {
+    let tmp_file = &format!("{output_file}.new");
+    let _ = fs::remove_file(tmp_file).await;
+    try_exec("wget", &["-O", tmp_file, seed_url]).await?;
+
+    fs::rename(tmp_file, output_file)
+        .await
+        .map_err(|e| format_err!("seed rename failed: {e}"))?;
+
+    Ok(())
+}
+
+async fn mount_system(cfg: &dkl::Config, bs_dir: &str) {
+    let mem_dir = "/mem";
+    mount("none", mem_dir, "tmpfs", Some("size=512m")).await;
+
+    let layers_dir = &format!("{mem_dir}/layers");
+    let mut lower_dir = String::new();
+
+    for layer in &cfg.layers {
+        let src = if layer == "modules" {
+            "/modules.sqfs"
+        } else {
+            &format!("{bs_dir}/{layer}.fs")
+        };
+        let tgt = &format!("{mem_dir}/{layer}.fs");
+        retry(async || {
+            info!("copying layer {layer} from {src}");
+            fs::copy(src, tgt).await?;
+            Ok(())
+        })
+        .await;
+
+        let layer_dir = &format!("{layers_dir}/{layer}");
+        mount(tgt, layer_dir, "squashfs", None).await;
+
+        if !lower_dir.is_empty() {
+            lower_dir.push(':');
+        }
+        lower_dir.push_str(&layer_dir);
+    }
+
+    let upper_dir = &format!("{mem_dir}/upper");
+    let work_dir = &format!("{mem_dir}/work");
+
+    retry_or_ignore(async || {
+        fs::create_dir_all(upper_dir).await?;
+        fs::create_dir_all(work_dir).await?;
+        Ok(())
+    })
+    .await;
+
+    mount(
+        "none",
+        "/system",
+        "overlay",
+        Some(&format!(
+            "lowerdir={lower_dir},upperdir={upper_dir},workdir={work_dir}"
+        )),
+    )
+    .await;
+
+    // make root rshared (default in systemd, required by Kubernetes 1.10+)
+    // equivalent to "mount --make-rshared /"
+    // see kernel's Documentation/sharedsubtree.txt (search rshared)
+    retry_or_ignore(async || {
+        use nix::mount::MsFlags as M;
+        const NONE: Option<&str> = None;
+        nix::mount::mount(NONE, "/system", NONE, M::MS_SHARED | M::MS_REC, NONE)?;
+        Ok(())
+    })
+    .await;
+}
+
+fn chroot(root: &str, path: &str) -> String {
+    format!("{root}/{}", path.trim_start_matches(|c| c == '/'))
+}
+
+async fn apply_files(files: &[dkl::File], root: &str) -> Result<()> {
+    for file in files {
+        let path = chroot(root, &file.path);
+        let path = Path::new(&path);
+
+        if let Some(parent) = path.parent() {
+            fs::create_dir_all(parent).await?;
+        }
+
+        use crate::dkl::FileKind as K;
+        match &file.kind {
+            K::Content(content) => fs::write(path, content.as_bytes()).await?,
+            K::Dir(true) => fs::create_dir(path).await?,
+            K::Dir(false) => {} // shouldn't happen, but semantic is to ignore
+            K::Symlink(tgt) => fs::symlink(tgt, path).await?,
+        }
+
+        match file.kind {
+            K::Symlink(_) => warn!("{}: can set perms on symlink", file.path),
+            _ => set_perms(path, file.mode).await?,
+        }
+
+        info!("created {}", file.path);
+    }
+
+    Ok(())
+}
+
+async fn set_perms(path: impl AsRef<Path>, mode: Option<u32>) -> std::io::Result<()> {
+    if let Some(mode) = mode.filter(|m| *m != 0) {
+        use std::os::unix::fs::PermissionsExt;
+        let mode = std::fs::Permissions::from_mode(mode);
+        fs::set_permissions(path, mode).await?;
+    }
+    Ok(())
+}
+
+async fn apply_groups(groups: &[dkl::Group], root: &str) {
+    for group in groups {
+        let mut args = vec![root, "groupadd", "-r"];
+
+        let gid = group.gid.map(|s| s.to_string());
+        if let Some(gid) = gid.as_ref() {
+            args.extend(&["-g", gid]);
+        }
+
+        args.push(group.name.as_str());
+
+        exec("chroot", &args).await;
+    }
+}
+
+async fn apply_users(users: &[dkl::User], root: &str) {
+    for user in users {
+        let mut args = vec![root, "useradd", "-r"];
+
+        let uid = user.uid.map(|s| s.to_string());
+        if let Some(uid) = uid.as_ref() {
+            args.extend(&["-u", uid]);
+        }
+        let gid = user.gid.map(|s| s.to_string());
+        if let Some(gid) = gid.as_ref() {
+            args.extend(&["-g", gid]);
+        }
+
+        args.push(user.name.as_str());
+
+        exec("chroot", &args).await;
+    }
+}
+
+async fn mount_filesystems(mounts: &[dkl::Mount], root: &str) {
+    for m in mounts {
+        let path = chroot(root, &m.path);
+
+        mount(
+            &m.dev,
+            &path,
+            m.r#type.as_ref().map_or("", |v| v.as_str()),
+            m.options
+                .as_ref()
+                .filter(|v| !v.is_empty())
+                .map(|s| s.as_str()),
+        )
+        .await;
+    }
+}
+
+async fn setup_root_user(user: &dkl::RootUser, root: &str) -> Result<()> {
+    if let Some(pw_hash) = user.password_hash.as_ref().filter(|v| !v.is_empty()) {
+        set_user_password("root", &pw_hash, root).await?;
+    }
+
+    let mut authorized_keys = Vec::new();
+    for ak in &user.authorized_keys {
+        authorized_keys.extend(ak.as_bytes());
+        authorized_keys.push(b'\n');
+    }
+
+    let ssh_dir = &chroot(root, "root/.ssh");
+    fs::create_dir_all(ssh_dir)
+        .await
+        .map_err(|e| format_err!("mkdir -p {ssh_dir} failed: {e}"))?;
+    set_perms(ssh_dir, Some(0o700))
+        .await
+        .map_err(|e| format_err!("chmod {ssh_dir} failed: {e}"))?;
+
+    let ak_path = &format!("{ssh_dir}/authorized_keys");
+    fs::write(ak_path, authorized_keys)
+        .await
+        .map_err(|e| format_err!("write {ak_path} failed: {e}"))?;
+
+    Ok(())
+}
+
+async fn set_user_password(user: &str, password_hash: &str, root: &str) -> Result<()> {
+    info!("setting password for {user}");
+
+    let user = user.as_bytes();
+    let password_hash = password_hash.as_bytes();
+
+    let mut buf = Vec::new();
+
+    let pw_file = &chroot(root, "etc/shadow");
+    let rd = fs::File::open(pw_file)
+        .await
+        .map_err(|e| format_err!("open {pw_file} failed: {e}"))?;
+    let mut rd = BufReader::new(rd);
+
+    let mut line = Vec::new();
+    while (rd.read_until(b'\n', &mut line).await)
+        .map_err(|e| format_err!("read {pw_file} failed: {e}"))?
+        != 0
+    {
+        let mut split: Vec<_> = line.split(|c| *c == b':').collect();
+        if split.len() > 2 && split[0] == user {
+            split[1] = password_hash;
+            buf.extend(split.join(&b':'));
+        } else {
+            buf.extend(&line);
+        }
+        line.clear();
+    }
+
+    fs::write(pw_file, buf).await?;
+    Ok(())
+}

src/cmd/init/dmcrypt.rs

@@ -0,0 +1,142 @@
+use eyre::{format_err, Result};
+use log::{error, info};
+use std::collections::BTreeSet as Set;
+use std::process::Stdio;
+use tokio::io::AsyncWriteExt;
+use tokio::process::Command;
+use tokio::sync::Mutex;
+
+use super::{retry_or_ignore, USED_DEVS};
+use crate::blockdev::is_uninitialized;
+use crate::bootstrap::config::{CryptDev, DevFilter};
+use crate::fs::walk_dir;
+use crate::input;
+
+pub async fn setup(devs: &[CryptDev]) {
+    if devs.is_empty() {
+        return;
+    }
+
+    let mut used_devs = USED_DEVS.lock().await;
+
+    // CryptDev.name that have a least one assignment done
+    let mut done = Set::new();
+
+    // dmcrypt devices opened here
+    let mut done_crypt = Set::new();
+
+    retry_or_ignore(async || {
+        let all_devs = walk_dir("/dev").await;
+
+        for dev in devs {
+            let mut mappings = find_dev(dev, &all_devs);
+            mappings.retain(|(_, dev_path)| !used_devs.contains(dev_path));
+
+            if mappings.is_empty() && !dev.optional() && !done.contains(&dev.name) {
+                return Err(format_err!("no device found for crypt dev {}", dev.name));
+            }
+
+            for (crypt_dev, dev_path) in mappings {
+                if done_crypt.contains(&crypt_dev) {
+                    continue;
+                }
+
+                info!("crypt dev {crypt_dev}: using {dev_path}");
+
+                crypt_open(&crypt_dev, &dev_path).await?;
+
+                done_crypt.insert(crypt_dev);
+                used_devs.insert(dev_path);
+                done.insert(&dev.name);
+            }
+        }
+
+        Ok(())
+    })
+    .await;
+}
+
+static PREV_PW: Mutex<String> = Mutex::const_new(String::new());
+
+async fn crypt_open(crypt_dev: &str, dev_path: &str) -> Result<()> {
+    'open_loop: loop {
+        let mut prev_pw = PREV_PW.lock().await;
+        let prompt = if prev_pw.is_empty() {
+            format!("crypt password for {crypt_dev}? ")
+        } else {
+            format!("crypt password for {crypt_dev} (enter = reuse previous)? ")
+        };
+
+        let mut pw = input::read_password(prompt).await;
+        if pw.is_empty() {
+            pw = prev_pw.clone();
+        }
+        if pw.is_empty() {
+            error!("empty password provided!");
+            continue;
+        }
+
+        *prev_pw = pw.clone();
+
+        if cryptsetup(&pw, ["open", dev_path, crypt_dev]).await? {
+            return Ok(());
+        }
+
+        error!("crypt open {crypt_dev} from {dev_path} failed");
+
+        if is_uninitialized(dev_path).await? {
+            // we can format the device
+            match input::read_choice(["[f]ormat", "[r]etry", "[i]gnore"]).await {
+                'r' => continue 'open_loop,
+                'i' => return Ok(()),
+                'f' => {
+                    if !cryptsetup(&pw, ["luksFormat", dev_path]).await? {
+                        return Err(format_err!("cryptsetup luksFormat failed"));
+                    }
+                    if !cryptsetup(&pw, ["open", dev_path, crypt_dev]).await? {
+                        return Err(format_err!("open after format failed"));
+                    }
+                    return Ok(());
+                }
+                _ => unreachable!(),
+            }
+        } else {
+            // device looks initialized, don't allow format
+            match input::read_choice(["[r]etry", "[i]gnore"]).await {
+                'r' => continue 'open_loop,
+                'i' => return Ok(()),
+                _ => unreachable!(),
+            }
+        }
+    }
+}
+
+async fn cryptsetup<const N: usize>(pw: &str, args: [&str; N]) -> Result<bool> {
+    let mut child = Command::new("cryptsetup")
+        .args(args)
+        .arg("--key-file=-")
+        .stdin(Stdio::piped())
+        .spawn()?;
+
+    (child.stdin.as_mut().unwrap())
+        .write_all(pw.as_bytes())
+        .await?;
+
+    Ok(child.wait().await?.success())
+}
+
+fn find_dev(dev: &CryptDev, all_devs: &[String]) -> Vec<(String, String)> {
+    let dev_name = &dev.name;
+    match dev.filter {
+        DevFilter::Dev(ref path) => (all_devs.iter())
+            .filter(|dev_path| dev_path == &path)
+            .map(|dev_path| (dev.name.clone(), dev_path.clone()))
+            .collect(),
+        DevFilter::Prefix(ref prefix) => (all_devs.iter())
+            .filter_map(|path| {
+                let suffix = path.strip_prefix(prefix)?;
+                Some((format!("{dev_name}{suffix}"), path.clone()))
+            })
+            .collect(),
+    }
+}

src/cmd/init/lvm.rs

@@ -0,0 +1,208 @@
+use eyre::{format_err, Result};
+use log::{error, info};
+use tokio::process::Command;
+
+use super::{exec, retry, retry_or_ignore, USED_DEVS};
+use crate::bootstrap::config::{Config, Filesystem, LvSize, LvmLV, LvmVG, TAKE_ALL};
+use crate::fs::walk_dir;
+use crate::{blockdev, lvm};
+
+pub async fn setup(cfg: &Config) {
+    if cfg.lvm.is_empty() {
+        info!("no LVM VG configured");
+        return;
+    }
+
+    exec("pvscan", &[]).await;
+    exec("vgscan", &["--mknodes"]).await;
+
+    for vg in &cfg.lvm {
+        retry_or_ignore(async || setup_vg(vg).await).await
+    }
+
+    let lvs = retry(lvm::lvs).await;
+
+    for vg in &cfg.lvm {
+        let vg_name = vg.name.as_str();
+
+        for lv in &vg.lvs {
+            let lv_name = lv.name.as_str();
+
+            if (lvs.iter()).any(|lv| lv.equal_name(vg_name, lv_name)) {
+                info!("LVM LV {vg_name}/{lv_name} exists");
+            } else {
+                retry_or_ignore(async || setup_lv(&vg, &lv).await).await;
+            }
+        }
+    }
+
+    exec("vgchange", &["--sysinit", "-a", "ly"]).await;
+
+    for vg in &cfg.lvm {
+        for lv in &vg.lvs {
+            retry_or_ignore(async || format_lv(&vg, &lv).await).await;
+        }
+    }
+}
+
+async fn setup_vg(vg: &LvmVG) -> Result<()> {
+    let vg_name = vg.name.as_str();
+
+    let pvs = retry(lvm::pvs).await;
+
+    let mut dev_done = pvs.iter().filter(|pv| pv.vg_name == vg.name).count();
+    let dev_needed = vg.pvs.n;
+    macro_rules! missing_count {
+        () => {
+            (dev_needed as usize) - dev_done
+        };
+    }
+
+    if dev_needed == TAKE_ALL {
+        if dev_done == 0 {
+            info!("setting up LVM VG {vg_name} using all matching devices");
+        } else {
+            // in "take all" mode, don't extend as existing vg at boot
+            info!("LVM VG {vg_name} exists");
+            return Ok(());
+        }
+    } else if dev_done >= (dev_needed as usize) {
+        info!("LVM VG {vg_name} exists with enough devices");
+        return Ok(()); // already set up
+    } else {
+        info!("setting up LVM VG {vg_name} ({dev_done}/{dev_needed} devices configured)");
+    }
+
+    let regexps: Vec<regex::Regex> = (vg.pvs.regexps.iter())
+        .filter_map(|re_str| {
+            (re_str.parse())
+                .inspect_err(|e| error!("invalid regex ignored: {re_str:?}: {e}"))
+                .ok()
+        })
+        .collect();
+
+    let mut used_devs = USED_DEVS.lock().await;
+
+    let matching_devs = (walk_dir("/dev").await.into_iter())
+        .filter(|path| !used_devs.contains(path.as_str()))
+        .filter(|path| regexps.iter().any(|re| re.is_match(path)));
+
+    let devs: Vec<_> = if dev_needed == TAKE_ALL {
+        matching_devs.collect()
+    } else {
+        matching_devs.take(missing_count!()).collect()
+    };
+
+    let cmd = if dev_done == 0 {
+        "vgcreate"
+    } else {
+        "vgextend"
+    };
+    let status = (Command::new(cmd).arg(vg_name).args(&devs))
+        .status()
+        .await?;
+    if !status.success() {
+        return Err(format_err!("{cmd} failed: {status}"));
+    }
+
+    dev_done += devs.len();
+    used_devs.extend(devs);
+
+    if dev_needed != TAKE_ALL && dev_done < (dev_needed as usize) {
+        return Err(format_err!(
+            "LVM VG {vg_name} needs {} more device(s)",
+            missing_count!()
+        ));
+    }
+
+    Ok(())
+}
+
+async fn setup_lv(vg: &LvmVG, lv: &LvmLV) -> Result<()> {
+    let name = format!("{}/{}", vg.name, lv.name);
+    info!("creating LV {name}");
+
+    let mut cmd = Command::new("lvcreate");
+    cmd.arg(&vg.name);
+    cmd.args(&["--name", &lv.name]);
+
+    match &lv.size {
+        LvSize::Size(sz) => cmd.args(&["-L", sz]),
+        LvSize::Extents(sz) => cmd.args(&["-l", sz]),
+    };
+
+    let raid = lv.raid.as_ref().unwrap_or(&vg.defaults.raid);
+
+    if let Some(mirrors) = raid.mirrors {
+        cmd.args(&["--mirrors", &mirrors.to_string()]);
+    }
+    if let Some(stripes) = raid.stripes {
+        cmd.args(&["--stripes", &stripes.to_string()]);
+    }
+
+    let status = cmd.status().await?;
+    if !status.success() {
+        return Err(format_err!("lvcreate failed: {status}"));
+    }
+
+    Ok(())
+}
+
+async fn format_lv(vg: &LvmVG, lv: &LvmLV) -> Result<()> {
+    let name = &format!("{}/{}", vg.name, lv.name);
+    let dev = &format!("/dev/{name}");
+
+    if !blockdev::is_uninitialized(&dev).await? {
+        info!("{dev} looks initialized");
+        return Ok(());
+    }
+
+    let fs = lv.fs.as_ref().unwrap_or(&vg.defaults.fs);
+    info!("initializing {} filesystem on {dev}", fs.fstype());
+
+    let mkfs = format!("mkfs.{}", fs.fstype());
+
+    let mut cmd = Command::new(&mkfs);
+
+    // filesystem specific flags
+    match fs {
+        Filesystem::Ext4 => {
+            cmd.arg("-F");
+        }
+        Filesystem::Btrfs | Filesystem::Xfs => {
+            cmd.arg("-f");
+        }
+        &Filesystem::Other(_) => {}
+    }
+
+    cmd.arg(dev);
+
+    let mut child = match cmd.spawn() {
+        Ok(v) => v,
+        Err(e) => {
+            // try simple fixes
+            match fs {
+                Filesystem::Xfs => install_package("xfsprogs").await?,
+                Filesystem::Btrfs => install_package("btrs-progs").await?,
+                _ => Err(format_err!("{mkfs} failed: {e}"))?,
+            }
+            cmd.spawn().map_err(|e| format_err!("{mkfs} failed: {e}"))?
+        }
+    };
+
+    let status = child.wait().await?;
+    if !status.success() {
+        return Err(format_err!("{mkfs} failed: {status}"));
+    }
+
+    Ok(())
+}
+
+async fn install_package(pkg: &str) -> Result<()> {
+    let status = Command::new("apk").arg("add").arg(pkg).status().await?;
+    if status.success() {
+        Ok(())
+    } else {
+        Err(format_err!("failed to install package {pkg}: {status}"))
+    }
+}

src/cmd/init/networks.rs

@@ -0,0 +1,94 @@
+use itertools::Itertools;
+use log::{info, warn};
+use std::collections::BTreeSet as Set;
+use tokio::process::Command;
+
+use super::{format_err, retry_or_ignore, Config, Result};
+use crate::{
+    bootstrap::config,
+    udev,
+    utils::{select_n_by_regex, NameAliases},
+};
+
+pub async fn setup(cfg: &Config) {
+    if cfg.networks.is_empty() {
+        warn!("no networks configured");
+        return;
+    }
+
+    let mut assigned = Set::new();
+
+    for net in &cfg.networks {
+        retry_or_ignore(async || setup_network(net, &mut assigned).await).await;
+    }
+}
+
+async fn setup_network(net: &config::Network, assigned: &mut Set<String>) -> Result<()> {
+    info!("setting up network {}", net.name);
+
+    let netdevs = get_interfaces()?
+        .filter(|dev| !assigned.contains(dev.name()))
+        .collect::<Vec<_>>();
+
+    for dev in &netdevs {
+        info!(
+            "- available network device: {}, aliases [{}]",
+            dev.name(),
+            dev.aliases().join(", ")
+        );
+    }
+
+    let mut cmd = Command::new("ash");
+    cmd.arg("-c");
+    cmd.arg(&net.script);
+
+    let mut selected = Vec::new();
+
+    for iface in &net.interfaces {
+        let var = &iface.var;
+
+        let netdevs = netdevs.iter().filter(|na| !assigned.contains(na.name()));
+        let if_names = select_n_by_regex(iface.n, &iface.regexps, netdevs);
+
+        if if_names.is_empty() {
+            return Err(format_err!("- no interface match for {var:?}"));
+        }
+
+        let value = if_names.join(" ");
+        info!("- {var}={value}");
+        cmd.env(var, value);
+
+        selected.extend(if_names);
+    }
+
+    info!("- running script");
+    let status = cmd.status().await?;
+    if !status.success() {
+        return Err(format_err!("setup script failed: {status}"));
+    }
+
+    assigned.extend(selected);
+    Ok(())
+}
+
+fn get_interfaces() -> Result<impl Iterator<Item = NameAliases>> {
+    Ok(udev::get_devices("net")?.into_iter().map(|dev| {
+        let mut na = NameAliases::new(dev.sysname().to_string());
+
+        for (property, value) in dev.properties() {
+            if [
+                "INTERFACE",
+                "ID_NET_NAME",
+                "ID_NET_NAME_PATH",
+                "ID_NET_NAME_MAC",
+                "ID_NET_NAME_SLOT",
+            ]
+            .contains(&property)
+            {
+                na.push(value.to_string());
+            }
+        }
+
+        na
+    }))
+}

src/cmd/init/sshd.rs

@@ -0,0 +1,91 @@
+use log::{info, warn};
+use std::fs;
+use std::io::Write;
+use std::os::unix::fs::PermissionsExt;
+use std::process::Stdio;
+use tokio::net;
+use tokio::process::Command;
+
+use super::{retry_or_ignore, Config};
+
+pub async fn start(cfg: &Config) {
+    retry_or_ignore(async || {
+        info!("ssh: writing authorized keys");
+
+        let ssh_dir = "/root/.ssh";
+        let authorized_keys = format!("{ssh_dir}/authorized_keys");
+
+        fs::create_dir_all(ssh_dir)?;
+        fs::set_permissions(ssh_dir, fs::Permissions::from_mode(0o700))?;
+
+        let mut ak = Vec::new();
+
+        for auth in &cfg.auths {
+            writeln!(ak, "{} {}", auth.ssh_key, auth.name)?;
+        }
+
+        fs::write(authorized_keys, ak)?;
+        Ok(())
+    })
+    .await;
+
+    retry_or_ignore(async || {
+        let mut sshd_args = Vec::new();
+
+        sshd_args.extend(["-i", "-E", "/var/log/sshd.log"]);
+
+        for key_path in cfg.ssh.keys.iter() {
+            if !fs::exists(key_path).is_ok_and(|b| b) {
+                info!("ssh: host key not found (ignored): {key_path}");
+                continue;
+            }
+            sshd_args.extend(["-h", key_path]);
+        }
+
+        let sshd_args = sshd_args.into_iter().map(String::from).collect();
+
+        // don't pre-start sshd as it should rarely be useful at this stage, use inetd-style.
+        let listen_addr = cfg.ssh.listen.clone();
+        info!("ssh: starting listener on {listen_addr}");
+
+        let listener = net::TcpListener::bind(listen_addr).await?;
+
+        tokio::spawn(handle_ssh_connections(listener, sshd_args));
+
+        Ok(())
+    })
+    .await;
+}
+
+async fn handle_ssh_connections(listener: net::TcpListener, sshd_args: Vec<String>) {
+    loop {
+        let (stream, remote) = match listener.accept().await {
+            Ok(v) => v,
+            Err(e) => {
+                warn!("ssh: listener stopped: {e}");
+                return;
+            }
+        };
+
+        info!("ssh: new connection from {remote}");
+
+        use std::os::unix::io::{AsRawFd, FromRawFd};
+        let fd = stream.as_raw_fd();
+
+        let mut cmd = Command::new("/usr/sbin/sshd");
+        cmd.args(&sshd_args);
+
+        cmd.stdin(unsafe { Stdio::from_raw_fd(fd) });
+        cmd.stdout(unsafe { Stdio::from_raw_fd(fd) });
+        cmd.stderr(Stdio::null());
+
+        match cmd.spawn() {
+            Ok(mut child) => {
+                tokio::spawn(async move { child.wait().await });
+            }
+            Err(e) => {
+                warn!("ssh: failed to start server: {e}");
+            }
+        }
+    }
+}

src/cmd/init_input.rs

@@ -0,0 +1,11 @@
+use crate::input;
+
+pub async fn run() {
+    tokio::spawn(async {
+        if let Err(e) = input::forward_requests_from_socket().await {
+            eprintln!("failed to forwards requests from socket: {e}");
+            std::process::exit(1);
+        }
+    });
+    input::answer_requests_from_stdin().await;
+}

src/cmd/version.rs

@@ -0,0 +1,12 @@
+pub fn run() {
+    println!("{}", version_string());
+}
+
+pub fn version_string() -> String {
+    format!(
+        "Direktil {} v{} (git commit {})",
+        env!("CARGO_PKG_NAME"),
+        env!("CARGO_PKG_VERSION"),
+        env!("GIT_COMMIT")
+    )
+}