feat(harmony): add TFTP server functionality (#10)

Introduce a new module and interface for serving files via TFTP in the HAClusterTopology structure. This includes adding the necessary dependencies, creating the `TftpServer` trait, implementing it where appropriate, and integrating its usage within the topology struct.

Reviewed-on: NationTech/harmony#10
Co-authored-by: Jean-Gabriel Gill-Couture <jg@nationtech.io>
Co-committed-by: Jean-Gabriel Gill-Couture <jg@nationtech.io>

harmony-rs/opnsense-config/Cargo.toml

@@ -18,6 +18,8 @@ opnsense-config-xml = { path = "../opnsense-config-xml" }
 chrono = "0.4.38"
 russh-sftp = "2.0.6"
 serde_json = "1.0.133"
+tokio-util = "0.7.13"
+tokio-stream = "0.1.17"
 
 [dev-dependencies]
 pretty_assertions = "1.4.1"

harmony-rs/opnsense-config/src/config/config.rs

@@ -3,7 +3,7 @@ use std::{sync::Arc, time::Duration};
 use crate::{
     config::{SshConfigManager, SshCredentials, SshOPNSenseShell},
     error::Error,
-    modules::{dhcp::DhcpConfig, dns::DnsConfig, load_balancer::LoadBalancerConfig},
+    modules::{dhcp::DhcpConfig, dns::DnsConfig, load_balancer::LoadBalancerConfig, tftp::TftpConfig},
 };
 use log::{info, trace};
 use opnsense_config_xml::OPNsense;
@@ -38,10 +38,18 @@ impl Config {
         DnsConfig::new(&mut self.opnsense, self.shell.clone())
     }
 
+    pub fn tftp(&mut self) -> TftpConfig {
+        TftpConfig::new(&mut self.opnsense, self.shell.clone())
+    }
+
     pub fn load_balancer(&mut self) -> LoadBalancerConfig {
         LoadBalancerConfig::new(&mut self.opnsense, self.shell.clone())
     }
 
+    pub async fn upload_files(&self, source: &str, destination: &str) ->  Result<String, Error> {
+        self.shell.upload_folder(source, destination).await
+    }
+
     pub async fn install_package(&mut self, package_name: &str) -> Result<(), Error> {
         info!("Installing opnsense package {package_name}");
         let output = self.shell

harmony-rs/opnsense-config/src/config/manager/ssh.rs

@@ -27,19 +27,25 @@ impl SshConfigManager {
         let ts = chrono::Utc::now();
         let backup_filename = format!("config-{}-harmony.xml", ts.format("%s%.3f"));
 
-        self.opnsense_shell.exec(&format!("cp /conf/config.xml /conf/backup/{}", backup_filename))
+        self.opnsense_shell
+            .exec(&format!(
+                "cp /conf/config.xml /conf/backup/{}",
+                backup_filename
+            ))
             .await
     }
 
     async fn move_to_live_config(&self, new_config_path: &str) -> Result<String, Error> {
         info!("Overwriting OPNSense /conf/config.xml with {new_config_path}");
-        self.opnsense_shell.exec(&format!("mv {new_config_path} /conf/config.xml"))
+        self.opnsense_shell
+            .exec(&format!("mv {new_config_path} /conf/config.xml"))
             .await
     }
 
     async fn reload_all_services(&self) -> Result<String, Error> {
         info!("Reloading all opnsense services");
-        self.opnsense_shell.exec(&format!("configctl service reload all"))
+        self.opnsense_shell
+            .exec(&format!("configctl service reload all"))
             .await
     }
 }

harmony-rs/opnsense-config/src/config/shell/mod.rs

@@ -9,6 +9,7 @@ use crate::Error;
 pub trait OPNsenseShell: std::fmt::Debug + Send + Sync {
     async fn exec(&self, command: &str) -> Result<String, Error>;
     async fn write_content_to_temp_file(&self, content: &str) -> Result<String, Error>;
+    async fn upload_folder(&self, source: &str, destination: &str) -> Result<String, Error>;
 }
 
 #[cfg(test)]
@@ -24,4 +25,7 @@ impl OPNsenseShell for DummyOPNSenseShell {
     async fn write_content_to_temp_file(&self, _content: &str) -> Result<String, Error> {
         unimplemented!("This is a dummy implementation");
     }
+    async fn upload_folder(&self, _source: &str, _destination: &str) -> Result<String, Error> {
+        unimplemented!("This is a dummy implementation");
+    }
 }

harmony-rs/opnsense-config/src/config/shell/ssh.rs

@@ -3,6 +3,7 @@ use std::{
     sync::Arc,
     time::{SystemTime, UNIX_EPOCH},
 };
+use tokio_stream::StreamExt;
 
 use async_trait::async_trait;
 use log::{debug, info};
@@ -11,12 +12,15 @@ use russh::{
     Channel,
 };
 use russh_keys::key;
-use russh_sftp::client::SftpSession;
-use tokio::io::AsyncWriteExt;
+use russh_sftp::{client::SftpSession, protocol::OpenFlags};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
 use crate::{config::SshCredentials, Error};
 
 use super::OPNsenseShell;
+use tokio::fs::read_dir;
+use tokio::fs::File;
+use tokio_util::codec::{BytesCodec, FramedRead};
 
 #[derive(Debug)]
 pub struct SshOPNSenseShell {
@@ -54,6 +58,72 @@ impl OPNsenseShell for SshOPNSenseShell {
 
         Ok(temp_filename)
     }
+
+    async fn upload_folder(&self, source: &str, destination: &str) -> Result<String, Error> {
+        let channel = self.get_ssh_channel().await?;
+        channel
+            .request_subsystem(true, "sftp")
+            .await
+            .expect("Should request sftp subsystem");
+        let sftp = SftpSession::new(channel.into_stream())
+            .await
+            .expect("Should acquire sftp subsystem");
+
+        if !sftp.try_exists(destination).await? {
+            info!("Creating remote directory {destination}");
+            sftp.create_dir(destination).await?;
+        }
+
+        info!("Reading local directory {source}");
+        let mut entries = read_dir(source).await?;
+        while let Some(entry) = entries.next_entry().await? {
+            info!(
+                "Checking directory entry {}",
+                entry
+                    .path()
+                    .to_str()
+                    .expect("Directory entry should have a path : {entry:?}")
+            );
+            if entry.file_type().await?.is_file() {
+                debug!("Got a file");
+                let local_path = entry.path();
+                debug!("path {local_path:?}");
+                let file_name = local_path.file_name().unwrap().to_string_lossy();
+                let remote_path = format!("{}/{}", destination, file_name);
+                info!(
+                    "Uploading local file {} to remote {}",
+                    local_path.to_str().unwrap_or_default(),
+                    remote_path
+                );
+
+                debug!("Creating file {remote_path:?}");
+                let mut remote_file = sftp.create(remote_path.as_str()).await?;
+
+                debug!("Writing file {remote_path:?}");
+                let local_file = File::open(&local_path).await?;
+                let mut reader = FramedRead::new(local_file, BytesCodec::new());
+
+                while let Some(result) = reader.next().await {
+                    match result {
+                        Ok(bytes) => {
+                            if !bytes.is_empty() {
+                                remote_file.write(&bytes).await?;
+                            }
+                        }
+                        Err(e) => todo!("Error unhandled {e}"),
+                    };
+                }
+            } else if entry.file_type().await?.is_dir() {
+                let sub_source = entry.path();
+                let sub_destination =
+                    format!("{}/{}", destination, entry.file_name().to_string_lossy());
+                self.upload_folder(sub_source.to_str().unwrap(), &sub_destination)
+                    .await?;
+            }
+        }
+
+        Ok(destination.to_string())
+    }
 }
 
 impl SshOPNSenseShell {
@@ -80,6 +150,7 @@ impl SshOPNSenseShell {
     }
 
     pub fn new(host: (IpAddr, u16), credentials: SshCredentials, ssh_config: Arc<Config>) -> Self {
+        info!("Initializing SshOPNSenseShell on host {host:?}");
         Self {
             host,
             credentials,

harmony-rs/opnsense-config/src/error.rs

@@ -6,6 +6,8 @@ pub enum Error {
     Xml(String),
     #[error("SSH error: {0}")]
     Ssh(#[from] russh::Error),
+    #[error("SSH Client error: {0}")]
+    SftpClient(#[from] russh_sftp::client::error::Error),
     #[error("Command failed : {0}")]
     Command(String),
     #[error("I/O error: {0}")]

harmony-rs/opnsense-config/src/modules/mod.rs

@@ -1,3 +1,4 @@
 pub mod dhcp;
 pub mod dns;
 pub mod load_balancer;
+pub mod tftp;

harmony-rs/opnsense-config/src/modules/tftp.rs

@@ -0,0 +1,48 @@
+use std::sync::Arc;
+
+use opnsense_config_xml::{OPNsense, Tftp};
+
+use crate::{config::OPNsenseShell, Error};
+
+pub struct TftpConfig<'a> {
+    opnsense: &'a mut OPNsense,
+    opnsense_shell: Arc<dyn OPNsenseShell>,
+}
+
+impl<'a> TftpConfig<'a> {
+    pub fn new(opnsense: &'a mut OPNsense, opnsense_shell: Arc<dyn OPNsenseShell>) -> Self {
+        Self {
+            opnsense,
+            opnsense_shell,
+        }
+    }
+
+    pub fn get_full_config(&self) -> &Option<Tftp> {
+        &self.opnsense.opnsense.tftp
+    }
+
+    fn with_tftp<F, R>(&mut self, f: F) -> R
+    where
+        F: FnOnce(&mut Tftp) -> R,
+    {
+        match &mut self.opnsense.opnsense.tftp.as_mut() {
+            Some(tftp) => f(tftp),
+            None => unimplemented!("Accessing tftp config is not supported when not available yet"),
+        }
+    }
+
+    pub fn enable(&mut self, enabled: bool) {
+        self.with_tftp(|tftp| tftp.general.enabled = enabled as u8);
+    }
+
+    pub fn listen_ip(&mut self, ip: &str) {
+        self.with_tftp(|tftp| tftp.general.listen = ip.to_string());
+    }
+
+    pub async fn reload_restart(&self) -> Result<(), Error> {
+        self.opnsense_shell.exec("configctl tftp stop").await?;
+        self.opnsense_shell.exec("configctl template reload OPNsense/Tftp").await?;
+        self.opnsense_shell.exec("configctl tftp start").await?;
+        Ok(())
+    }
+}