--wip-- [skip ci]

Cargo.lock

@@ -124,6 +124,70 @@ version = "1.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
 
+[[package]]
+name = "axum"
+version = "0.8.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
+dependencies = [
+ "axum-core",
+ "axum-macros",
+ "bytes",
+ "form_urlencoded",
+ "futures-util",
+ "http",
+ "http-body",
+ "http-body-util",
+ "hyper",
+ "hyper-util",
+ "itoa",
+ "matchit",
+ "memchr",
+ "mime",
+ "percent-encoding",
+ "pin-project-lite",
+ "serde_core",
+ "serde_json",
+ "serde_path_to_error",
+ "serde_urlencoded",
+ "sync_wrapper",
+ "tokio",
+ "tower",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
+[[package]]
+name = "axum-core"
+version = "0.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "http",
+ "http-body",
+ "http-body-util",
+ "mime",
+ "pin-project-lite",
+ "sync_wrapper",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
+[[package]]
+name = "axum-macros"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7aa268c23bfbbd2c4363b9cd302a4f504fb2a9dfe7e3451d66f35dd392e20aca"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
 [[package]]
 name = "base64"
 version = "0.23.1"
@@ -293,6 +357,7 @@ name = "crete"
 version = "0.1.3"
 dependencies = [
  "async-walkdir",
+ "axum",
  "byteorder",
  "clap",
  "clap_complete",
@@ -312,6 +377,7 @@ dependencies = [
  "serde_yaml",
  "thiserror",
  "tokio",
+ "udp_sas_async",
 ]
 
 [[package]]
@@ -860,6 +926,12 @@ version = "1.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
 
+[[package]]
+name = "httpdate"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
+
 [[package]]
 name = "hyper"
 version = "1.11.1"
@@ -873,6 +945,7 @@ dependencies = [
  "http",
  "http-body",
  "httparse",
+ "httpdate",
  "itoa",
  "pin-project-lite",
  "smallvec",
@@ -1211,6 +1284,12 @@ version = "0.4.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
 
+[[package]]
+name = "matchit"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3"
+
 [[package]]
 name = "memchr"
 version = "2.8.3"
@@ -1232,6 +1311,12 @@ version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5449c8c750f1a07ea702bbd212bd999fceece9b3d1508b17023b3e174583124b"
 
+[[package]]
+name = "mime"
+version = "0.3.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
+
 [[package]]
 name = "minijinja"
 version = "2.24.0"
@@ -1375,6 +1460,16 @@ dependencies = [
  "syn 2.0.119",
 ]
 
+[[package]]
+name = "os_socketaddr"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fa149f3ec09cadeb41968390fcd71e6ea038a9a373d1d4eb070f4d9484d4b244"
+dependencies = [
+ "libc",
+ "winapi",
+]
+
 [[package]]
 name = "parking"
 version = "2.2.1"
@@ -1790,6 +1885,29 @@ dependencies = [
  "zmij",
 ]
 
+[[package]]
+name = "serde_path_to_error"
+version = "0.1.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457"
+dependencies = [
+ "itoa",
+ "serde",
+ "serde_core",
+]
+
+[[package]]
+name = "serde_urlencoded"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd"
+dependencies = [
+ "form_urlencoded",
+ "itoa",
+ "ryu",
+ "serde",
+]
+
 [[package]]
 name = "serde_yaml"
 version = "0.9.34+deprecated"
@@ -2031,6 +2149,7 @@ dependencies = [
  "tokio",
  "tower-layer",
  "tower-service",
+ "tracing",
 ]
 
 [[package]]
@@ -2069,6 +2188,7 @@ version = "0.1.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
 dependencies = [
+ "log",
  "pin-project-lite",
  "tracing-core",
 ]
@@ -2088,6 +2208,18 @@ version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
 
+[[package]]
+name = "udp_sas_async"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "073390c0d7b1682a5f044db2f769b0564568f741b9a7e23b8494b79fc868d491"
+dependencies = [
+ "cc",
+ "libc",
+ "os_socketaddr",
+ "tokio",
+]
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.26"
@@ -2230,6 +2362,22 @@ dependencies = [
  "wasm-bindgen",
 ]
 
+[[package]]
+name = "winapi"
+version = "0.3.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419"
+dependencies = [
+ "winapi-i686-pc-windows-gnu",
+ "winapi-x86_64-pc-windows-gnu",
+]
+
+[[package]]
+name = "winapi-i686-pc-windows-gnu"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6"
+
 [[package]]
 name = "winapi-util"
 version = "0.1.11"
@@ -2239,6 +2387,12 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "winapi-x86_64-pc-windows-gnu"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
+
 [[package]]
 name = "windows-link"
 version = "0.2.1"

Cargo.toml

@@ -36,6 +36,8 @@ serde_yaml = "0.9.34"
 thiserror = "2.0.21"
 tokio = { version = "1.52.1", features = ["full"] }
 ipxe = { path = "./ipxe" }
+axum = { version = "0.8.9", features = ["macros"] }
+udp_sas_async = { version = "0.2.0", features = ["async_tokio"] }
 
 [lib]
 test = false

ipxe/build.rs

@@ -23,6 +23,7 @@ fn main() {
                 .get()
                 .to_string(),
         ])
+        .arg("DOWNLOAD_PROTO_HTTP=1")
         .arg(format!(
             "EMBED={}",
             manifest_dir

ipxe/src/embed.ipxe

@@ -1,5 +1,4 @@
 #!ipxe
 
 dhcp
-
+chain boot.ipxe
-echo "Hello, World!"

src/boot.ipxe

@@ -0,0 +1,20 @@
+{% set httpUrl = "http://192.168.1.1:8000" -%}
+#!ipxe
+echo Starting ${serial}
+
+:start
+goto node_${serial} || exit
+# Default behavior (non install mode) is to exit iPXE script
+
+{% for cluster in clusters%}
+{% for _, node in cluster.nodes|items %}
+{%- if node.install.serial -%}
+# {{ cluster.name }}/{{ node.hostname }}
+:node_{{ node.install.serial }}
+imgfree
+kernel https://pxe.factory.talos.dev/image/{{ node.schematic }}/v{{ cluster.version.talos }}/kernel-{{ node.arch }} {{ node.kernelArgs|join(" ") }} {% if node.install.auto %}talos.config={{httpUrl}}/talos/{{cluster.name}}/{{node.hostname}}.yaml{% endif +%}
+initrd https://pxe.factory.talos.dev/image/{{ node.schematic }}/v{{ cluster.version.talos }}/initramfs-{{ node.arch }}.xz
+boot
+{% endif %}
+{% endfor %}
+{% endfor %}

src/environment.rs

@@ -25,46 +25,50 @@ fn apply_netmask(
     Ok((*ip & *netmask).to_string())
 }
 
+pub fn configure_environment(env: &mut Environment) {
+    // Configure jinja
+    env.set_trim_blocks(true);
+    env.set_lstrip_blocks(true);
+    env.set_undefined_behavior(minijinja::UndefinedBehavior::Strict);
+    env.set_auto_escape_callback(|_| AutoEscape::None);
+
+    // Add filters
+    env.add_filter("to_prefix", |netmask: ViaDeserialize<Ipv4Addr>| {
+        let mask = netmask.to_bits();
+        let prefix = mask.leading_ones();
+
+        if mask.checked_shl(prefix).unwrap_or(0) == 0 {
+            Ok(prefix as u8)
+        } else {
+            Err(minijinja::Error::new(
+                minijinja::ErrorKind::InvalidOperation,
+                "invalid IP prefix length",
+            ))
+        }
+    });
+
+    env.add_function("apply_netmask", apply_netmask);
+
+    // Helper function for getting the path to kubeconfig files
+    env.add_filter("kubeconfig", move |names: Vec<String>| {
+        names
+            .iter()
+            .map(|name| {
+                get_configs_path()
+                    .join(name)
+                    .join("kubeconfig")
+                    .to_string_lossy()
+                    .to_string()
+            })
+            .collect::<Vec<_>>()
+    });
+}
+
 impl<'a> PathEnvironment<'a> {
     pub fn new(path: &Path) -> Self {
         let mut env = Environment::new();
 
-        // Configure jinja
+        configure_environment(&mut env);
-        env.set_trim_blocks(true);
-        env.set_lstrip_blocks(true);
-        env.set_undefined_behavior(minijinja::UndefinedBehavior::Strict);
-        env.set_auto_escape_callback(|_| AutoEscape::None);
-
-        // Add filters
-        env.add_filter("to_prefix", |netmask: ViaDeserialize<Ipv4Addr>| {
-            let mask = netmask.to_bits();
-            let prefix = mask.leading_ones();
-
-            if mask.checked_shl(prefix).unwrap_or(0) == 0 {
-                Ok(prefix as u8)
-            } else {
-                Err(minijinja::Error::new(
-                    minijinja::ErrorKind::InvalidOperation,
-                    "invalid IP prefix length",
-                ))
-            }
-        });
-
-        env.add_function("apply_netmask", apply_netmask);
-
-        // Helper function for getting the path to kubeconfig files
-        env.add_filter("kubeconfig", move |names: Vec<String>| {
-            names
-                .iter()
-                .map(|name| {
-                    get_configs_path()
-                        .join(name)
-                        .join("kubeconfig")
-                        .to_string_lossy()
-                        .to_string()
-                })
-                .collect::<Vec<_>>()
-        });
 
         // Add path loader
         env.set_loader(path_loader(&path));

src/main.rs

@@ -1,19 +1,25 @@
 mod cli;
 
+use std::path::PathBuf;
 use std::pin::pin;
 
+use axum::extract::Request;
+use axum::response::IntoResponse;
+use axum::routing::get;
+use axum::{Router, debug_handler};
 use clap::{CommandFactory, Parser};
 use clap_complete::{Shell, generate as generate_complete};
 use crete::cluster::Cluster;
-use crete::environment::PathEnvironment;
+use crete::environment::{PathEnvironment, configure_environment};
 use crete::talosconfig::{get_current_context, set_context};
 use crete::tftp::ServerError;
 use crete::{get_configs_path, get_repo_path, set_repo_path, tftp};
 use futures::StreamExt;
 use futures::future::join_all;
 use ipxe::IPXE_DIR;
-use minijinja::context;
+use minijinja::{Environment, context};
 use regex::Regex;
+use reqwest::{StatusCode, header};
 use thiserror::Error;
 
 use crate::cli::{Cli, Commands, GlobalOpts};
@@ -93,6 +99,92 @@ async fn generate(opts: &GlobalOpts) -> Result<(), Error> {
     Ok(())
 }
 
+#[debug_handler]
+async fn handler(req: Request) -> impl IntoResponse {
+    println!("{req:?}");
+    let host = req
+        .headers()
+        .iter()
+        .find(|(name, _)| header::HOST.eq(name))
+        .map(|(_, value)| value)
+        .unwrap()
+        .to_str()
+        .unwrap();
+
+    println!("{host}");
+
+    let Some(path) = req.uri().path().strip_prefix("/") else {
+        return StatusCode::BAD_REQUEST.into_response();
+    };
+
+    let Some((file, content_type)) = get_file(path).await else {
+        return StatusCode::NOT_FOUND.into_response();
+    };
+
+    let path = PathBuf::from(path);
+    let filename = path.file_name().unwrap();
+    let filename = filename.to_str().unwrap();
+
+    let content_type = [(header::CONTENT_TYPE, content_type)];
+    let disposition = [(
+        header::CONTENT_DISPOSITION,
+        format!("inline; filename=\"{filename}\""),
+    )];
+
+    (content_type, disposition, file).into_response()
+}
+
+// TODO: Make this a proper async function
+fn get_file(filename: impl Into<String>) -> impl Future<Output = Option<(Vec<u8>, &'static str)>> {
+    // TODO: Make this a static that is initialized once
+    let re =
+        Regex::new(r"^talos/(?<cluster>.+)/(?<node>.+)\.ya?ml$").expect("regex should be valid");
+
+    let filename: String = filename.into();
+    let re = re.clone();
+
+    async move {
+        const IPXE_PATTERN: &str = "ipxe/";
+        if filename == format!("{IPXE_PATTERN}boot.ipxe") {
+            let mut env = Environment::new();
+
+            configure_environment(&mut env);
+
+            let template = env
+                .template_from_str(include_str!("./boot.ipxe"))
+                .expect("Template should be valid");
+
+            let clusters = Cluster::get_all().await;
+            let content = template
+                .render(context! {
+                    clusters => clusters.values().collect::<Vec<_>>(),
+                })
+                .unwrap();
+
+            Some((content.into(), "text/plain; charset=utf-8"))
+        } else if let Some(filename) = filename.strip_prefix(IPXE_PATTERN) {
+            Some((
+                IPXE_DIR.get_file(filename)?.contents().to_owned(),
+                "application/octet-stream",
+            ))
+        } else if let Some(caps) = re.captures(&filename) {
+            let cluster = &caps["cluster"];
+            let node = &caps["node"];
+
+            let clusters = Cluster::get_all().await;
+            let cluster = clusters.get(cluster)?;
+
+            let node = cluster.node(node)?;
+
+            let config = node.config(cluster).await;
+
+            Some((config.into(), "application/yaml; charset=utf-8"))
+        } else {
+            None
+        }
+    }
+}
+
 #[tokio::main]
 async fn main() -> Result<(), Error> {
     let cli = Cli::parse();
@@ -108,35 +200,17 @@ async fn main() -> Result<(), Error> {
             &mut std::io::stdout(),
         ),
         Commands::Serve => {
-            let re = Regex::new(r"talos/(?<cluster>.+)/(?<node>.+)\.ya?ml")
+            tokio::spawn(async {
-                .expect("regex should be valid");
+                let app = Router::new().fallback_service(get(handler));
+                let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
+                axum::serve(listener, app).await.unwrap();
+            });
 
             tftp::serve(move |filename| {
                 let filename = filename.to_owned();
-                let re = re.clone();
+                async move { get_file(filename).await.map(|(file, _)| file) }
-
-                async move {
-                    const IPXE_PATTERN: &str = "ipxe/";
-                    if let Some(filename) = filename.strip_prefix(IPXE_PATTERN) {
-                        Some(IPXE_DIR.get_file(filename)?.contents().to_owned())
-                    } else if let Some(caps) = re.captures(&filename) {
-                        let cluster = &caps["cluster"];
-                        let node = &caps["node"];
-
-                        let clusters = Cluster::get_all().await;
-                        let cluster = clusters.get(cluster)?;
-
-                        let node = cluster.node(node)?;
-
-                        let config = node.config(cluster).await;
-
-                        Some(config.into())
-                    } else {
-                        None
-                    }
-                }
             })
-            .await?
+            .await?;
         }
     };
 

src/tftp/packet.rs

@@ -3,7 +3,7 @@ use std::net::SocketAddr;
 
 use byteorder::{NetworkEndian, ReadBytesExt, WriteBytesExt};
 use num_enum::{IntoPrimitive, TryFromPrimitive};
-use tokio::net::UdpSocket;
+use udp_sas_async::tokio::UdpSocketSas;
 
 use crate::tftp::Error;
 
@@ -309,16 +309,15 @@ impl Packet {
         Ok(())
     }
 
-    pub async fn recv_from(socket: &UdpSocket) -> Result<(Self, SocketAddr), Error> {
+    pub async fn recv_from(socket: &UdpSocketSas) -> Result<(Self, SocketAddr), Error> {
         let mut buf = Vec::with_capacity(PACKET_SIZE);
-        let (_, src) = socket.recv_buf_from(&mut buf).await?;
+        let (_, src) = socket.recv_from(&mut buf).await?;
-
         let packet = buf.try_into()?;
 
         Ok((packet, src))
     }
 
-    pub async fn send_to(&self, socket: &UdpSocket, src: &SocketAddr) -> Result<(), Error> {
+    pub async fn send_to(&self, socket: &UdpSocketSas, src: SocketAddr) -> Result<(), Error> {
         let buf: Vec<_> = self.try_into()?;
         socket.send_to(&buf, src).await?;
 

src/tftp/server.rs

@@ -1,9 +1,9 @@
 use std::io;
-use std::net::SocketAddr;
+use std::net::{IpAddr, SocketAddr};
 use std::time::Duration;
 
 use thiserror::Error;
-use tokio::net::UdpSocket;
+use udp_sas_async::tokio::UdpSocketSas;
 
 use crate::tftp::packet::{ErrorCode, Mode};
 use crate::tftp::{self, Packet};
@@ -29,23 +29,25 @@ pub enum Error {
 }
 
 pub async fn handle_connection<F, Fut>(
-    src: SocketAddr,
     packet: Packet,
+    src: SocketAddr,
+    dest: IpAddr,
     callback: F,
 ) -> Result<(), Error>
 where
     F: Fn(&str) -> Fut,
     Fut: Future<Output = Option<Vec<u8>>>,
 {
-    let socket = UdpSocket::bind("0.0.0.0:0").await?;
+    let socket = UdpSocketSas::bind(
-    println!(
+        format!("{dest}:0")
-        "{src}: Connecting from {}",
+            .parse()
-        socket.local_addr().expect("Socket is bound")
+            .expect("Address should be valid"),
-    );
+    )?;
+    println!("{src}: Connecting from {}", socket.local_addr());
 
     let Packet::Read(request) = packet else {
         Packet::error(ErrorCode::NotDefined, "Operation not supported")
-            .send_to(&socket, &src)
+            .send_to(&socket, src)
             .await?;
 
         return Err(Error::UnsupportedOperation(packet));
@@ -56,7 +58,7 @@ where
 
     let Some(data) = callback(filename).await else {
         Packet::error(ErrorCode::FileNotFound, "Unknown file")
-            .send_to(&socket, &src)
+            .send_to(&socket, src)
             .await?;
 
         return Err(Error::FileNotFound(filename.into()));
@@ -72,7 +74,7 @@ where
                 ErrorCode::NotDefined,
                 format!("Mode {mode:?} is not supported"),
             )
-            .send_to(&socket, &src)
+            .send_to(&socket, src)
             .await?;
 
             return Err(Error::UnsupportedMode(mode));
@@ -88,10 +90,9 @@ where
         // Send the first chunk
         if let Some((index, chunk)) = chunk {
             Packet::data((index + 1) as u16, chunk)
-                .send_to(&socket, &src)
+                .send_to(&socket, src)
                 .await?;
         }
-
         tokio::select! {
             recv = Packet::recv_from(&socket) => {
                 timeout = 0;
@@ -101,7 +102,7 @@ where
                 // really know how to test it properly so we don't support it.
                 if new_src != src {
                     Packet::error(ErrorCode::Illegal, "Source changed")
-                        .send_to(&socket, &src)
+                        .send_to(&socket, src)
                         .await?;
 
                     return Err(Error::SourceChanged(new_src));
@@ -122,7 +123,7 @@ where
                     }
                     _ => {
                         Packet::error(ErrorCode::Illegal, "Unexpected packet")
-                            .send_to(&socket, &src)
+                            .send_to(&socket, src)
                             .await?;
 
 
@@ -150,16 +151,21 @@ where
     F: Fn(&str) -> Fut + Clone + Send + 'static,
     Fut: Future<Output = Option<Vec<u8>>> + Send + 'static,
 {
-    let socket = UdpSocket::bind("0.0.0.0:69").await?;
+    let socket = UdpSocketSas::bind("0.0.0.0:69".parse().expect("Address should be valid"))?;
 
     println!("Server started");
     loop {
+        let mut buf = Vec::with_capacity(0);
+        let (bytes, src, dest) = socket.recv_sas(&mut buf).await?;
+        println!("{src} -> {dest}");
+        println!("{bytes}, {}", buf.len());
+
         let (packet, src) = Packet::recv_from(&socket).await?;
 
         let callback = callback.clone();
         tokio::spawn({
             async move {
-                if let Err(err) = handle_connection(src, packet, callback).await {
+                if let Err(err) = handle_connection(packet, src, dest, callback).await {
                     println!("{src}: {err}")
                 }
             }