use human_units for Duration params

.gitignore

@@ -1,8 +1,12 @@
 /target
 /dls
 /modd.conf
-/m1_bootstrap-config
-/config.yaml
 /dist
 /dkl
 /tmp
+
+/config.yaml
+
+# dls assets
+/cluster_*_*
+/host_*_*

Cargo.toml

@@ -12,17 +12,21 @@ codegen-units = 1
 
 [dependencies]
 async-compression = { version = "0.4.27", features = ["tokio", "zstd"] }
+base32 = "0.5.1"
 bytes = "1.10.1"
-chrono = { version = "0.4.41", default-features = false, features = ["now"] }
+chrono = { version = "0.4.41", default-features = false, features = ["clock", "now"] }
 clap = { version = "4.5.40", features = ["derive", "env"] }
 clap_complete = { version = "4.5.54", features = ["unstable-dynamic"] }
 env_logger = "0.11.8"
 eyre = "0.6.12"
+fastrand = "2.3.0"
 futures = "0.3.31"
 futures-util = "0.3.31"
 glob = "0.3.2"
 hex = "0.4.3"
+human-units = "0.5.3"
 log = "0.4.27"
+lz4 = "1.28.1"
 nix = { version = "0.30.1", features = ["user"] }
 openssl = "0.10.73"
 page_size = "0.6.0"

src/bin/dkl.rs

@@ -1,6 +1,8 @@
 use clap::{CommandFactory, Parser, Subcommand};
 use eyre::{format_err, Result};
+use human_units::Duration;
 use log::{debug, error};
+use std::net::SocketAddr;
 use tokio::fs;
 
 #[derive(Parser)]
@@ -65,6 +67,17 @@ enum Command {
         #[arg(long, default_value = "/")]
         chroot: std::path::PathBuf,
     },
+    Proxy {
+        #[arg(long, short = 'l')]
+        listen: Vec<SocketAddr>,
+        targets: Vec<SocketAddr>,
+        /// target polling interval
+        #[arg(long, default_value = "30s")]
+        poll: Duration,
+        /// connect or check timeout
+        #[arg(long, default_value = "5s")]
+        timeout: Duration,
+    },
 }
 
 #[tokio::main(flavor = "current_thread")]
@@ -126,6 +139,20 @@ async fn main() -> Result<()> {
             .install(layer, version)
             .await
         }
+        C::Proxy {
+            listen,
+            targets,
+            poll,
+            timeout,
+        } => Ok(dkl::proxy::Proxy {
+            listen_addrs: listen,
+            targets,
+            poll: poll.into(),
+            timeout: timeout.into(),
+        }
+        .run()
+        .await
+        .map(|_| ())?),
     }
 }
 

src/bin/dls.rs

@@ -1,14 +1,18 @@
+use bytes::Bytes;
 use clap::{CommandFactory, Parser, Subcommand};
-use eyre::{Result, format_err};
+use eyre::format_err;
+use futures_util::Stream;
 use futures_util::StreamExt;
-use tokio::io::AsyncWriteExt;
+use std::time::{Duration, SystemTime};
+use tokio::fs;
+use tokio::io::{AsyncWrite, AsyncWriteExt};
 
 use dkl::dls;
 
 #[derive(Parser)]
 #[command()]
 struct Cli {
-    #[arg(long, default_value = "http://[::1]:7606")]
+    #[arg(long, default_value = "http://[::1]:7606", env = "DLS_URL")]
     dls: String,
 
     #[command(subcommand)]
@@ -25,9 +29,36 @@ enum Command {
     },
     Hosts,
     Host {
+        #[arg(short = 'o', long)]
+        out: Option<String>,
         host: String,
         asset: Option<String>,
     },
+    #[command(subcommand)]
+    DlSet(DlSet),
+}
+
+#[derive(Subcommand)]
+enum DlSet {
+    Sign {
+        #[arg(short = 'e', long, default_value = "1d")]
+        expiry: String,
+        #[arg(value_parser = parse_download_set_item)]
+        items: Vec<dls::DownloadSetItem>,
+    },
+    Show {
+        #[arg(env = "DLS_DLSET")]
+        signed_set: String,
+    },
+    Fetch {
+        #[arg(long, env = "DLS_DLSET")]
+        signed_set: String,
+        #[arg(short = 'o', long)]
+        out: Option<String>,
+        kind: String,
+        name: String,
+        asset: String,
+    },
 }
 
 #[derive(Subcommand)]
@@ -62,7 +93,7 @@ enum ClusterCommand {
 }
 
 #[tokio::main(flavor = "current_thread")]
-async fn main() -> Result<()> {
+async fn main() -> eyre::Result<()> {
     clap_complete::CompleteEnv::with_factory(Cli::command).complete();
 
     let cli = Cli::parse();
@@ -125,44 +156,110 @@ async fn main() -> Result<()> {
             }
         }
         C::Hosts => write_json(&dls.hosts().await?),
-        C::Host { host, asset } => {
+        C::Host { out, host, asset } => {
             let host_name = host.clone();
             let host = dls.host(host);
             match asset {
                 None => write_json(&host.config().await?),
                 Some(asset) => {
-                    let mut stream = host.asset(&asset).await?;
-
-                    let out_path = format!("{host_name}_{asset}");
-                    eprintln!("writing {host_name} asset {asset} to {out_path}");
-
-                    let out = tokio::fs::File::options()
-                        .mode(0o600)
-                        .write(true)
-                        .create(true)
-                        .truncate(true)
-                        .open(out_path)
-                        .await?;
-                    let mut out = tokio::io::BufWriter::new(out);
-
-                    let mut n = 0u64;
-                    while let Some(chunk) = stream.next().await {
-                        let chunk = chunk?;
-                        n += chunk.len() as u64;
-                        eprint!("wrote {n} bytes\r");
-                        out.write_all(&chunk).await?;
-                    }
-                    eprintln!();
-
-                    out.flush().await?;
+                    let stream = host.asset(&asset).await?;
+                    let mut out = create_asset_file(out, "host", &host_name, &asset).await?;
+                    copy_stream(stream, &mut out).await?;
                 }
             }
         }
+        C::DlSet(set) => match set {
+            DlSet::Sign { expiry, items } => {
+                let req = dls::DownloadSetReq { expiry, items };
+                let signed = dls.sign_dl_set(&req).await?;
+                println!("{signed}");
+            }
+            DlSet::Show { signed_set } => {
+                let raw = base32::decode(base32::Alphabet::Rfc4648 { padding: false }, &signed_set)
+                    .ok_or(format_err!("invalid dlset"))?;
+
+                let sig_len = raw[0] as usize;
+                let (sig, data) = raw[1..].split_at(sig_len);
+                println!("signature: {}...", hex::encode(&sig[..16]));
+
+                let data = lz4::Decoder::new(data)?;
+                let data = std::io::read_to_string(data)?;
+
+                let (expiry, items) = data.split_once('|').ok_or(format_err!("invalid dlset"))?;
+                let expiry = i64::from_str_radix(expiry, 16)?;
+                let expiry = chrono::DateTime::from_timestamp(expiry, 0).unwrap();
+
+                println!("expires on {expiry}");
+
+                for item in items.split('|') {
+                    let mut parts = item.split(':');
+                    let Some(kind) = parts.next() else {
+                        continue;
+                    };
+                    let Some(name) = parts.next() else {
+                        continue;
+                    };
+                    for asset in parts {
+                        println!("- {kind} {name} {asset}");
+                    }
+                }
+            }
+            DlSet::Fetch {
+                signed_set,
+                out,
+                kind,
+                name,
+                asset,
+            } => {
+                let stream = dls.fetch_dl_set(&signed_set, &kind, &name, &asset).await?;
+                let mut out = create_asset_file(out, &kind, &name, &asset).await?;
+                copy_stream(stream, &mut out).await?;
+            }
+        },
     };
 
     Ok(())
 }
 
+async fn create_asset_file(
+    path: Option<String>,
+    kind: &str,
+    name: &str,
+    asset: &str,
+) -> std::io::Result<fs::File> {
+    let path = &path.unwrap_or(format!("{kind}_{name}_{asset}"));
+    eprintln!("writing {kind} {name} asset {asset} to {path}");
+    (fs::File::options().write(true).create(true).truncate(true))
+        .mode(0o600)
+        .open(path)
+        .await
+}
+
+async fn copy_stream(
+    mut stream: impl Stream<Item = reqwest::Result<Bytes>> + Unpin,
+    out: &mut (impl AsyncWrite + Unpin),
+) -> std::io::Result<()> {
+    let mut out = tokio::io::BufWriter::new(out);
+
+    let info_delay = Duration::from_secs(1);
+    let mut ts = SystemTime::now();
+
+    let mut n = 0u64;
+    while let Some(chunk) = stream.next().await {
+        let chunk = chunk.map_err(|e| std::io::Error::other(e))?;
+        n += chunk.len() as u64;
+        out.write_all(&chunk).await?;
+
+        if ts.elapsed().is_ok_and(|t| t >= info_delay) {
+            eprint!("wrote {n} bytes\r");
+            ts = SystemTime::now();
+        }
+    }
+    eprintln!("wrote {n} bytes");
+
+    out.flush().await
+}
+
 fn write_json<T: serde::ser::Serialize>(v: &T) {
     let data = serde_json::to_string_pretty(v).expect("value should serialize to json");
     println!("{data}");
@@ -172,6 +269,20 @@ fn write_raw(raw: &[u8]) {
     use std::io::Write;
 
     let mut out = std::io::stdout();
-    out.write(raw).expect("stdout write");
+    out.write_all(raw).expect("stdout write");
     out.flush().expect("stdout flush");
 }
+
+fn parse_download_set_item(s: &str) -> Result<dls::DownloadSetItem, std::io::Error> {
+    let err = |s: &str| std::io::Error::other(s);
+
+    let mut parts = s.split(':');
+
+    let item = dls::DownloadSetItem {
+        kind: parts.next().ok_or(err("no kind"))?.to_string(),
+        name: parts.next().ok_or(err("no name"))?.to_string(),
+        assets: parts.map(|p| p.to_string()).collect(),
+    };
+
+    Ok(item)
+}

src/bootstrap.rs

@@ -2,7 +2,7 @@ use std::collections::BTreeMap as Map;
 
 pub const TAKE_ALL: i16 = -1;
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct Config {
     pub anti_phishing_code: String,
 
@@ -14,20 +14,22 @@ pub struct Config {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub resolv_conf: Option<String>,
 
-    #[serde(default)]
+    #[serde(default, skip_serializing_if = "Map::is_empty")]
     pub vpns: Map<String, String>,
 
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub networks: Vec<Network>,
 
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub auths: Vec<Auth>,
     #[serde(default)]
     pub ssh: SSHServer,
 
-    #[serde(default)]
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub pre_lvm_crypt: Vec<CryptDev>,
-    #[serde(default)]
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub lvm: Vec<LvmVG>,
-    #[serde(default)]
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub crypt: Vec<CryptDev>,
 
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -36,7 +38,20 @@ pub struct Config {
     pub bootstrap: Bootstrap,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+impl Config {
+    pub fn new(bootstrap_dev: String) -> Self {
+        Self {
+            anti_phishing_code: "Direktil<3".into(),
+            bootstrap: Bootstrap {
+                dev: bootstrap_dev,
+                ..Default::default()
+            },
+            ..Default::default()
+        }
+    }
+}
+
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub struct Auth {
     pub name: String,
     #[serde(alias = "sshKey")]
@@ -46,18 +61,21 @@ pub struct Auth {
     pub password: Option<String>,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub struct Network {
     pub name: String,
     pub interfaces: Vec<NetworkInterface>,
     pub script: String,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub struct NetworkInterface {
     pub var: String,
     pub n: i16,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub regexps: Vec<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub udev: Option<UdevFilter>,
 }
 
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
@@ -74,7 +92,7 @@ impl Default for SSHServer {
     }
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub struct LvmVG {
     #[serde(alias = "vg")]
     pub name: String,
@@ -86,7 +104,7 @@ pub struct LvmVG {
     pub lvs: Vec<LvmLV>,
 }
 
-#[derive(Debug, Default, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct LvmLVDefaults {
     #[serde(default)]
     pub fs: Filesystem,
@@ -94,9 +112,10 @@ pub struct LvmLVDefaults {
     pub raid: Raid,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Default, Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "snake_case")]
 pub enum Filesystem {
+    #[default]
     Ext4,
     Xfs,
     Btrfs,
@@ -115,13 +134,7 @@ impl Filesystem {
     }
 }
 
-impl Default for Filesystem {
-    fn default() -> Self {
-        Filesystem::Ext4
-    }
-}
-
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub struct LvmLV {
     pub name: String,
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -132,61 +145,98 @@ pub struct LvmLV {
     pub size: LvSize,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "snake_case")]
 pub enum LvSize {
     Size(String),
     Extents(String),
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub struct LvmPV {
     pub n: i16,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub regexps: Vec<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub udev: Option<UdevFilter>,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 pub struct CryptDev {
     pub name: String,
-    #[serde(flatten)]
-    pub filter: DevFilter,
+    // hit the limit of enum representation here (flatten + enum variant case)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub dev: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub prefix: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub udev: Option<UdevFilter>,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub optional: Option<bool>,
 }
 impl CryptDev {
+    pub fn filter(&self) -> DevFilter<'_> {
+        if let Some(dev) = self.dev.as_deref() {
+            DevFilter::Dev(dev)
+        } else if let Some(prefix) = self.prefix.as_deref() {
+            DevFilter::Prefix(prefix)
+        } else if let Some(udev) = self.udev.as_ref() {
+            DevFilter::Udev(udev)
+        } else {
+            DevFilter::None
+        }
+    }
+
     pub fn optional(&self) -> bool {
-        self.optional.unwrap_or_else(|| self.filter.is_prefix())
+        self.optional.unwrap_or_else(|| match self.filter() {
+            DevFilter::None => true,
+            DevFilter::Dev(_) => false,
+            DevFilter::Prefix(_) => true,
+            DevFilter::Udev(_) => true,
+        })
     }
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[test]
+fn test_parse_crypt_dev() {
+    for s in [
+        "name: sys0\ndev: /dev/sda\n",
+        "name: crypt-\nprefix: /dev/sd\n",
+        "name: crypt-${name}\nudev: !glob [ DEVNAME, /dev/sd* ]\n",
+    ] {
+        let dev: CryptDev = serde_yaml::from_str(s).unwrap();
+        dev.filter();
+        dev.optional();
+    }
+}
+
+pub enum DevFilter<'t> {
+    None,
+    Dev(&'t str),
+    Prefix(&'t str),
+    Udev(&'t UdevFilter),
+}
+
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "snake_case")]
-pub enum DevFilter {
-    Dev(String),
-    Prefix(String),
-}
-impl DevFilter {
-    pub fn is_dev(&self) -> bool {
-        match self {
-            Self::Dev(_) => true,
-            _ => false,
-        }
-    }
-    pub fn is_prefix(&self) -> bool {
-        match self {
-            Self::Prefix(_) => true,
-            _ => false,
-        }
-    }
+pub enum UdevFilter {
+    Has(String),
+    Eq(String, String),
+    Glob(String, String),
+    And(Vec<UdevFilter>),
+    Or(Vec<UdevFilter>),
+    Not(Box<UdevFilter>),
 }
 
-#[derive(Debug, Default, Clone, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct Raid {
     pub mirrors: Option<u8>,
     pub stripes: Option<u8>,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct Bootstrap {
     pub dev: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub seed: Option<String>,
 }

src/dls.rs

@@ -33,7 +33,7 @@ impl Client {
         self.get_json("clusters").await
     }
 
-    pub fn cluster(&self, name: String) -> Cluster {
+    pub fn cluster(&self, name: String) -> Cluster<'_> {
         Cluster { dls: self, name }
     }
 
@@ -41,16 +41,30 @@ impl Client {
         self.get_json("hosts").await
     }
 
-    pub fn host(&self, name: String) -> Host {
+    pub fn host(&self, name: String) -> Host<'_> {
         Host { dls: self, name }
     }
 
-    pub async fn get_json<T: serde::de::DeserializeOwned>(&self, path: impl Display) -> Result<T> {
-        let req = self.get(&path)?.header("Accept", "application/json");
+    pub async fn sign_dl_set(&self, req: &DownloadSetReq) -> Result<String> {
+        let req = (self.req(Method::POST, "sign-download-set")?).json(req);
+        self.req_json(req).await
+    }
+    pub async fn fetch_dl_set(
+        &self,
+        signed_dlset: &str,
+        kind: &str,
+        name: &str,
+        asset: &str,
+    ) -> Result<impl Stream<Item = reqwest::Result<Bytes>>> {
+        let req = self.get(format!(
+            "public/download-set/{kind}/{name}/{asset}?set={signed_dlset}"
+        ))?;
         let resp = do_req(req, &self.token).await?;
+        Ok(resp.bytes_stream())
+    }
 
-        let body = resp.bytes().await.map_err(Error::Read)?;
-        serde_json::from_slice(&body).map_err(Error::Parse)
+    pub async fn get_json<T: serde::de::DeserializeOwned>(&self, path: impl Display) -> Result<T> {
+        self.req_json(self.get(&path)?).await
     }
     pub async fn get_bytes(&self, path: impl Display) -> Result<Vec<u8>> {
         let resp = do_req(self.get(&path)?, &self.token).await?;
@@ -60,6 +74,16 @@ impl Client {
         self.req(Method::GET, path)
     }
 
+    pub async fn req_json<T: serde::de::DeserializeOwned>(
+        &self,
+        req: reqwest::RequestBuilder,
+    ) -> Result<T> {
+        let req = req.header("Accept", "application/json");
+        let resp = do_req(req, &self.token).await?;
+
+        let body = resp.bytes().await.map_err(Error::Read)?;
+        serde_json::from_slice(&body).map_err(Error::Parse)
+    }
     pub fn req(&self, method: Method, path: impl Display) -> Result<reqwest::RequestBuilder> {
         let uri = format!("{}/{path}", self.base_url);
 
@@ -135,6 +159,30 @@ impl<'t> Host<'t> {
     }
 }
 
+#[derive(Default, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct Config {
+    #[serde(default, deserialize_with = "deserialize_null_as_default")]
+    pub clusters: Vec<ClusterConfig>,
+    #[serde(default, deserialize_with = "deserialize_null_as_default")]
+    pub hosts: Vec<HostConfig>,
+    #[serde(default, deserialize_with = "deserialize_null_as_default")]
+    pub host_templates: Vec<HostConfig>,
+    #[serde(default, rename = "SSLConfig")]
+    pub ssl_config: String,
+}
+
+// compensate for go's encoder pitfalls
+use serde::{Deserialize, Deserializer};
+fn deserialize_null_as_default<'de, D, T>(deserializer: D) -> std::result::Result<T, D::Error>
+where
+    T: Default + Deserialize<'de>,
+    D: Deserializer<'de>,
+{
+    let opt = Option::deserialize(deserializer)?;
+    Ok(opt.unwrap_or_default())
+}
+
 #[derive(serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "PascalCase")]
 pub struct ClusterConfig {
@@ -143,22 +191,36 @@ pub struct ClusterConfig {
     pub addons: String,
 }
 
-#[derive(serde::Deserialize, serde::Serialize)]
+#[derive(Default, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "PascalCase")]
 pub struct HostConfig {
     pub name: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
     pub cluster_name: Option<String>,
 
-    pub annotations: Map<String, String>,
-    pub bootstrap_config: String,
-    #[serde(rename = "IPXE")]
-    pub ipxe: Option<String>,
     #[serde(rename = "IPs")]
     pub ips: Vec<IpAddr>,
+
+    #[serde(default, skip_serializing_if = "Map::is_empty")]
+    pub labels: Map<String, String>,
+    #[serde(default, skip_serializing_if = "Map::is_empty")]
+    pub annotations: Map<String, String>,
+
+    #[serde(rename = "IPXE", skip_serializing_if = "Option::is_none")]
+    pub ipxe: Option<String>,
+
     pub initrd: String,
     pub kernel: String,
-    pub labels: Map<String, String>,
     pub versions: Map<String, String>,
+
+    /// initrd config template
+    pub bootstrap_config: String,
+    /// files to add to the final initrd config, with rendering
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub initrd_files: Vec<crate::File>,
+
+    /// system config template
+    pub config: String,
 }
 
 #[derive(serde::Deserialize, serde::Serialize)]
@@ -184,6 +246,23 @@ pub struct KubeSignReq {
     pub validity: Option<String>,
 }
 
+#[derive(serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct DownloadSetReq {
+    pub expiry: String,
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub items: Vec<DownloadSetItem>,
+}
+
+#[derive(Clone, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct DownloadSetItem {
+    pub kind: String,
+    pub name: String,
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub assets: Vec<String>,
+}
+
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 struct ServerError {
     #[serde(default)]

src/lib.rs

@@ -1,11 +1,12 @@
 pub mod apply;
+pub mod proxy;
 pub mod bootstrap;
 pub mod dls;
-pub mod logger;
 pub mod dynlay;
 pub mod fs;
+pub mod logger;
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct Config {
     pub layers: Vec<String>,
     pub root_user: RootUser,
@@ -19,18 +20,20 @@ pub struct Config {
     pub users: Vec<User>,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct RootUser {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub password_hash: Option<String>,
     pub authorized_keys: Vec<String>,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, Default, serde::Deserialize, serde::Serialize)]
 pub struct Mount {
-    pub r#type: Option<String>,
     pub dev: String,
     pub path: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub r#type: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub options: Option<String>,
 }
 
@@ -50,7 +53,7 @@ pub struct User {
     pub gid: Option<u32>,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
 pub struct File {
     pub path: String,
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -59,10 +62,21 @@ pub struct File {
     pub kind: FileKind,
 }
 
-#[derive(Debug, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, PartialEq, Eq, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "snake_case")]
 pub enum FileKind {
     Content(String),
     Symlink(String),
     Dir(bool),
 }
+
+// ------------------------------------------------------------------------
+
+impl Config {
+    pub fn has_file(&self, path: &str) -> bool {
+        self.files.iter().any(|f| f.path == path)
+    }
+    pub fn file(&self, path: &str) -> Option<&File> {
+        self.files.iter().find(|f| f.path == path)
+    }
+}

src/proxy.rs

@@ -0,0 +1,136 @@
+use log::{info, log_enabled, warn};
+use std::convert::Infallible;
+use std::io;
+use std::net::SocketAddr;
+use std::sync::atomic::{AtomicBool, Ordering::Relaxed};
+use std::time::Duration;
+use thiserror::Error;
+use tokio::net::{TcpListener, TcpStream};
+use tokio::time;
+
+pub struct Proxy {
+    pub listen_addrs: Vec<SocketAddr>,
+    pub targets: Vec<SocketAddr>,
+    pub poll: Duration,
+    pub timeout: Duration,
+}
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("failed to listen on {0}: {1}")]
+    ListenFailed(SocketAddr, std::io::Error),
+}
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+impl Proxy {
+    pub async fn run(self) -> Result<Infallible> {
+        let mut listeners = Vec::with_capacity(self.listen_addrs.len());
+        for addr in self.listen_addrs {
+            listeners.push(
+                TcpListener::bind(&addr)
+                    .await
+                    .map_err(|e| Error::ListenFailed(addr, e))?,
+            );
+            info!("listening on {addr}");
+        }
+
+        // all targets are initially ok (better land on a down one than just fail)
+        let targets: Vec<_> = (self.targets.into_iter())
+            .map(|addr| TargetStatus {
+                addr,
+                up: AtomicBool::new(true),
+                timeout: self.timeout,
+            })
+            .collect();
+
+        // the proxy runs forever -> using 'static is not a leak
+        let targets = targets.leak();
+
+        for listener in listeners {
+            tokio::spawn(proxy_listener(listener, targets));
+        }
+
+        check_targets(targets, self.poll).await
+    }
+}
+
+struct TargetStatus {
+    addr: SocketAddr,
+    up: AtomicBool,
+    timeout: Duration,
+}
+impl TargetStatus {
+    fn is_up(&self) -> bool {
+        self.up.load(Relaxed)
+    }
+
+    fn set_up(&self, is_up: bool) {
+        let prev = self.up.swap(is_up, Relaxed);
+        if prev != is_up {
+            if is_up {
+                info!("{} is up", self.addr);
+            } else {
+                warn!("{} is down", self.addr);
+            }
+        }
+    }
+
+    async fn connect(&self) -> io::Result<TcpStream> {
+        let r = match time::timeout(self.timeout, TcpStream::connect(self.addr)).await {
+            Ok(r) => r,
+            Err(e) => Err(io::Error::new(io::ErrorKind::TimedOut, e)),
+        };
+
+        self.set_up(r.is_ok());
+        r
+    }
+}
+
+async fn check_targets(targets: &'static [TargetStatus], poll: Duration) -> ! {
+    use tokio::time;
+    let mut poll_ticker = time::interval(poll);
+    poll_ticker.set_missed_tick_behavior(time::MissedTickBehavior::Skip);
+
+    loop {
+        poll_ticker.tick().await;
+
+        let mut tasks = tokio::task::JoinSet::new();
+
+        for target in targets {
+            tasks.spawn(target.connect());
+        }
+
+        tasks.join_all().await;
+
+        if log_enabled!(log::Level::Info) {
+            let mut infos = String::new();
+            for ts in targets.iter() {
+                infos.push_str(&format!("{} ", ts.addr));
+                infos.push_str(if ts.is_up() { "up   " } else { "down " });
+            }
+            info!("{infos}");
+        }
+    }
+}
+
+async fn proxy_listener(listener: TcpListener, targets: &'static [TargetStatus]) {
+    let mut rng = fastrand::Rng::new();
+
+    loop {
+        let mut active = Vec::with_capacity(targets.len());
+        let (mut src, _) = listener.accept().await.expect("listener.accept() failed");
+
+        active.extend((targets.iter().enumerate()).filter_map(|(i, ts)| ts.is_up().then_some(i)));
+        rng.shuffle(&mut active);
+
+        tokio::spawn(async move {
+            for i in active {
+                if let Ok(mut dst) = targets[i].connect().await {
+                    let _ = tokio::io::copy_bidirectional(&mut src, &mut dst).await;
+                    break;
+                }
+            }
+        });
+    }
+}

test-dls

@@ -18,3 +18,12 @@ $dls cluster cluster ssh-sign ~/.ssh/id_ed25519.pub
 $dls host m1 | jq '{Name, ClusterName, IPs}'
 $dls host m1 bootstrap-config
 
+export DLS_DLSET=$($dls dl-set sign --expiry 1d \
+    cluster:cluster:addons \
+    host:m1:kernel:initrd:bootstrap.tar \
+    host:m2:config:bootstrap-config:boot.vmdk)
+
+$dls dl-set show
+$dls dl-set fetch host m2 bootstrap-config
+rm host_m2_bootstrap-config
+