12 Commits
prod ... wip

8 changed files with 803 additions and 452 deletions

8
.gitignore vendored
View File

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

779
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@ -12,8 +12,9 @@ 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"
@ -23,6 +24,7 @@ futures-util = "0.3.31"
glob = "0.3.2"
hex = "0.4.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"

View File

@ -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)
}

View File

@ -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, 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,30 @@ 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(),
keymap: None,
modules: None,
resolv_conf: None,
vpns: Map::new(),
networks: vec![],
auths: vec![],
ssh: Default::default(),
pre_lvm_crypt: vec![],
lvm: vec![],
crypt: vec![],
signer_public_key: None,
bootstrap: Bootstrap {
dev: bootstrap_dev,
seed: None,
},
}
}
}
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
pub struct Auth {
pub name: String,
#[serde(alias = "sshKey")]
@ -46,18 +71,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 +102,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 +114,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 +122,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 +144,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 +155,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, serde::Deserialize, serde::Serialize)]
pub struct Bootstrap {
pub dev: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub seed: Option<String>,
}

View File

@ -5,6 +5,7 @@ use reqwest::Method;
use std::collections::BTreeMap as Map;
use std::fmt::Display;
use std::net::IpAddr;
use std::time::Duration;
pub struct Client {
base_url: String,
@ -33,7 +34,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 +42,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 +75,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);
@ -143,22 +168,33 @@ 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(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(skip_serializing_if = "Map::is_empty")]
pub labels: Map<String, String>,
#[serde(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>,
pub bootstrap_config: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub initrd_files: Vec<crate::File>,
pub config: String,
}
#[derive(serde::Deserialize, serde::Serialize)]
@ -184,6 +220,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)]
@ -253,3 +306,50 @@ pub enum Error {
#[error("response parsing failed: {0}")]
Parse(serde_json::Error),
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum File {
Static(crate::File),
Gen { path: String, from: ContentGen },
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ContentGen {
CaCrt(CaRef),
TlsKey(TlsRef),
TlsCrt {
key: TlsRef,
ca: CaRef,
profile: CertProfile,
},
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CaRef {
Global(String),
Cluster(String, String),
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TlsRef {
Cluster(String, String),
Host(String, String),
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CertProfile {
Client,
Server,
/// basicaly Client+Server
Peer,
Kube {
user: String,
group: String,
duration: Duration,
},
}

View File

@ -1,11 +1,11 @@
pub mod apply;
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 +19,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>,
}

View File

@ -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