--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",
@@ -310,6 +375,8 @@ dependencies = [
  "serde",
  "serde_json",
  "serde_yaml",
+ "socket-pktinfo",
+ "socket2",
  "thiserror",
  "tokio",
 ]
@@ -860,6 +927,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 +946,7 @@ dependencies = [
  "http",
  "http-body",
  "httparse",
+ "httpdate",
  "itoa",
  "pin-project-lite",
  "smallvec",
@@ -1211,6 +1285,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 +1312,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"
@@ -1790,6 +1876,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"
@@ -1837,6 +1946,17 @@ version = "1.16.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f9395f0f0eee849a9b707b2f06bb92a6a422090e2123bb2ef8e87a0e61892a8e"
 
+[[package]]
+name = "socket-pktinfo"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "612942246d0cc239cfd83af1dfd39be47f649208a3524e5e9da651910128e0ac"
+dependencies = [
+ "libc",
+ "socket2",
+ "windows-sys",
+]
+
 [[package]]
 name = "socket2"
 version = "0.6.5"
@@ -2031,6 +2151,7 @@ dependencies = [
  "tokio",
  "tower-layer",
  "tower-service",
+ "tracing",
 ]
 
 [[package]]
@@ -2069,6 +2190,7 @@ version = "0.1.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
 dependencies = [
+ "log",
  "pin-project-lite",
  "tracing-core",
 ]

Cargo.toml

@@ -36,6 +36,9 @@ 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"] }
+socket2 = "0.6.5"
+socket-pktinfo = "0.4.1"
 
 [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,19 @@
+#!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/cli.rs

@@ -1,3 +1,4 @@
+use std::net::IpAddr;
 use std::path::PathBuf;
 
 use clap::{Args, Parser, Subcommand};
@@ -50,5 +51,10 @@ pub enum Commands {
     /// Generate completions for your current shell
     ShellCompletions,
     /// Serve config files over tftp
-    Serve,
+    Serve {
+        #[arg(short, long, default_value = "3000")]
+        port: u16,
+        #[arg(long, default_value = "0.0.0.0")]
+        ip: IpAddr,
+    },
 }

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,27 @@
 mod cli;
 
+use std::net::SocketAddr;
+use std::path::PathBuf;
 use std::pin::pin;
 
+use axum::extract::Request;
+use axum::http::Uri;
+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 +101,99 @@ async fn generate(opts: &GlobalOpts) -> Result<(), Error> {
     Ok(())
 }
 
+#[debug_handler]
+async fn handler(req: Request) -> impl IntoResponse {
+    println!("{req:?}");
+    let uri: Uri = req
+        .headers()
+        .iter()
+        .find(|(name, _)| header::HOST.eq(name))
+        .map(|(_, value)| value)
+        .expect("Host header should exist")
+        .to_str()
+        .expect("Header value should be valid ASCII")
+        .parse()
+        .expect("Host should be valid Uri");
+
+    let host = uri.host().expect("Uri should have host");
+    let port = uri.port().map(|port| port.as_u16()).unwrap_or(80);
+
+    let Some(path) = req.uri().path().strip_prefix("/") else {
+        return StatusCode::BAD_REQUEST.into_response();
+    };
+
+    let Some((file, content_type)) = get_file(path, format!("http://{host}:{port}")).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 ToString,
+    http_url: impl ToString,
+) -> 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.to_string();
+    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<_>>(),
+                    httpUrl => http_url.to_string()
+                })
+                .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();
@@ -107,36 +208,25 @@ async fn main() -> Result<(), Error> {
             "crete",
             &mut std::io::stdout(),
         ),
-        Commands::Serve => {
+        Commands::Serve { port, ip } => {
-            let re = Regex::new(r"talos/(?<cluster>.+)/(?<node>.+)\.ya?ml")
+            tokio::spawn(async move {
-                .expect("regex should be valid");
+                let app = Router::new().fallback_service(get(handler));
+                let addr: SocketAddr = (ip, port).into();
+                let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
 
-            tftp::serve(move |filename| {
+                println!("Http server started on {addr}");
+                axum::serve(listener, app).await.unwrap();
+            });
+
+            tftp::serve(ip, move |filename, ip| {
                 let filename = filename.to_owned();
-                let re = re.clone();
-
                 async move {
-                    const IPXE_PATTERN: &str = "ipxe/";
+                    get_file(filename, format!("http://{ip}:{port}"))
-                    if let Some(filename) = filename.strip_prefix(IPXE_PATTERN) {
+                        .await
-                        Some(IPXE_DIR.get_file(filename)?.contents().to_owned())
+                        .map(|(file, _)| file)
-                    } 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

@@ -310,8 +310,8 @@ impl Packet {
     }
 
     pub async fn recv_from(socket: &UdpSocket) -> Result<(Self, SocketAddr), Error> {
-        let mut buf = Vec::with_capacity(PACKET_SIZE);
+        let mut buf = [0u8; PACKET_SIZE];
-        let (_, src) = socket.recv_buf_from(&mut buf).await?;
+        let (_, src) = socket.recv_from(&mut buf).await?;
 
         let packet = buf.try_into()?;
 
@@ -342,10 +342,12 @@ impl TryFrom<&Packet> for Vec<u8> {
     }
 }
 
-impl TryFrom<Vec<u8>> for Packet {
+impl TryFrom<[u8; PACKET_SIZE]> for Packet {
     type Error = Error;
 
-    fn try_from(buf: Vec<u8>) -> Result<Self, <Packet as TryFrom<Vec<u8>>>::Error> {
+    fn try_from(
+        buf: [u8; PACKET_SIZE],
+    ) -> Result<Self, <Packet as TryFrom<[u8; PACKET_SIZE]>>::Error> {
         let mut reader = BufReader::new(Cursor::new(buf));
 
         Packet::decode(&mut reader)

src/tftp/server.rs

@@ -1,11 +1,14 @@
 use std::io;
-use std::net::SocketAddr;
+use std::net::{IpAddr, SocketAddr};
 use std::time::Duration;
 
+use socket_pktinfo::{PktInfo, PktInfoUdpSocket};
+use socket2::Domain;
 use thiserror::Error;
+use tokio::io::unix::AsyncFd;
 use tokio::net::UdpSocket;
 
-use crate::tftp::packet::{ErrorCode, Mode};
+use crate::tftp::packet::{ErrorCode, Mode, PACKET_SIZE};
 use crate::tftp::{self, Packet};
 
 #[derive(Debug, Error)]
@@ -29,23 +32,25 @@ pub enum Error {
 }
 
 pub async fn handle_connection<F, Fut>(
-    src: SocketAddr,
+    info: PktInfo,
     packet: Packet,
     callback: F,
 ) -> Result<(), Error>
 where
-    F: Fn(&str) -> Fut,
+    F: Fn(&str, IpAddr) -> Fut,
     Fut: Future<Output = Option<Vec<u8>>>,
 {
-    let socket = UdpSocket::bind("0.0.0.0:0").await?;
+    let src = info.addr_src;
+    let socket = UdpSocket::bind((info.addr_dst, 0)).await?;
     println!(
-        "{src}: Connecting from {}",
+        "{}: Connecting back from {}",
-        socket.local_addr().expect("Socket is bound")
+        src,
+        socket.local_addr().expect("Local addr should be set")
     );
 
     let Packet::Read(request) = packet else {
         Packet::error(ErrorCode::NotDefined, "Operation not supported")
-            .send_to(&socket, &src)
+            .send_to(&socket, &info.addr_src)
             .await?;
 
         return Err(Error::UnsupportedOperation(packet));
@@ -54,7 +59,7 @@ where
     let filename = request.filename();
     println!("{src}: {filename} [{:?}]", request.mode());
 
-    let Some(data) = callback(filename).await else {
+    let Some(data) = callback(filename, info.addr_dst).await else {
         Packet::error(ErrorCode::FileNotFound, "Unknown file")
             .send_to(&socket, &src)
             .await?;
@@ -145,21 +150,36 @@ where
     Ok(())
 }
 
-pub async fn serve<F, Fut>(callback: F) -> Result<!, Error>
+pub async fn serve<F, Fut>(ip: IpAddr, callback: F) -> Result<!, Error>
 where
-    F: Fn(&str) -> Fut + Clone + Send + 'static,
+    F: Fn(&str, IpAddr) -> Fut + Clone + Send + 'static,
     Fut: Future<Output = Option<Vec<u8>>> + Send + 'static,
 {
-    let socket = UdpSocket::bind("0.0.0.0:69").await?;
+    let socket = PktInfoUdpSocket::new(Domain::IPV4)?;
+    socket.set_nonblocking(true)?;
 
-    println!("Server started");
+    let addr: SocketAddr = (ip, 69).into();
+    socket.bind(&addr.into())?;
+
+    let async_fd = AsyncFd::new(socket)?;
+    let mut buf = [0u8; PACKET_SIZE];
+
+    println!("Tftp server started on {addr}");
     loop {
-        let (packet, src) = Packet::recv_from(&socket).await?;
+        let mut guard = async_fd.readable().await?;
+
+        let (packet, info) = match guard.try_io(|inner| inner.get_ref().recv(&mut buf)) {
+            Ok(Ok((_, info))) => (buf.try_into()?, info),
+            Ok(Err(e)) => return Err(e)?,
+            // Socket was not ready
+            Err(_would_block) => continue,
+        };
 
         let callback = callback.clone();
+        let src = info.addr_src;
         tokio::spawn({
             async move {
-                if let Err(err) = handle_connection(src, packet, callback).await {
+                if let Err(err) = handle_connection(info, packet, callback).await {
                     println!("{src}: {err}")
                 }
             }