add ssh support

Cargo.toml

@@ -17,6 +17,9 @@ prost = "0.14"
 prost-types = "0.14"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+russh = "0.45"
+russh-keys = "0.45"
+async-trait = "0.1"
 
 [build-dependencies]
 prost-build = "0.14"

proto/agent.proto

@@ -174,6 +174,7 @@ message MikrotikDevice {
   string username = 3;
   string password = 4;
   bool use_ssl = 5;
+  uint32 ssh_port = 6;
 }
 
 message MikrotikCommand {

src/main.rs

@@ -1,6 +1,7 @@
 mod mikrotik;
 mod proto;
 mod snmp;
+mod ssh;
 mod version;
 mod websocket_client;
 

src/poller/executor.rs

@@ -21,6 +21,19 @@ use crate::snmp::SnmpClient;
 use crate::metrics::Timestamp;
 use log::{error, info, warn};
 
+/// Redact SNMP community string for logging, showing first 2 chars only
+fn redact_community(community: &str) -> String {
+    let len = community.len();
+    if len == 0 {
+        return "[redacted]".to_string();
+    }
+    if len <= 2 {
+        "**".to_string()
+    } else {
+        format!("{}***", &community[..2])
+    }
+}
+
 /// Executor handles polling individual pieces of equipment
 #[derive(Clone)]
 pub struct Executor {
@@ -43,9 +56,10 @@ impl Executor {
         }
 
         info!(
-            "Polling {} sensors for equipment: {}",
+            "Polling {} sensors for equipment: {} (community: {})",
             equipment.sensors.len(),
-            equipment.name
+            equipment.name,
+            redact_community(&equipment.snmp.community)
         );
 
         for sensor in &equipment.sensors {
@@ -113,9 +127,10 @@ impl Executor {
         }
 
         info!(
-            "Polling {} interfaces for equipment: {}",
+            "Polling {} interfaces for equipment: {} (community: {})",
             equipment.interfaces.len(),
-            equipment.name
+            equipment.name,
+            redact_community(&equipment.snmp.community)
         );
 
         for interface in &equipment.interfaces {
@@ -231,3 +246,34 @@ impl Executor {
             .ok_or_else(|| ExecutorError::Conversion("Could not convert value to i64".into()))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_redact_community_normal() {
+        assert_eq!(redact_community("public"), "pu***");
+    }
+
+    #[test]
+    fn test_redact_community_short() {
+        assert_eq!(redact_community("ab"), "**");
+        assert_eq!(redact_community("a"), "**");
+    }
+
+    #[test]
+    fn test_redact_community_empty() {
+        assert_eq!(redact_community(""), "[redacted]");
+    }
+
+    #[test]
+    fn test_redact_community_three_chars() {
+        assert_eq!(redact_community("abc"), "ab***");
+    }
+
+    #[test]
+    fn test_redact_community_long() {
+        assert_eq!(redact_community("mysecretcommunity"), "my***");
+    }
+}

src/ssh/client.rs

@@ -0,0 +1,145 @@
+use async_trait::async_trait;
+use russh::client;
+use russh_keys::key;
+use std::sync::Arc;
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum SshError {
+    #[error("SSH connection failed: {0}")]
+    ConnectionFailed(String),
+    #[error("SSH authentication failed")]
+    AuthenticationFailed,
+    #[error("SSH command execution failed: {0}")]
+    CommandFailed(String),
+    #[error("SSH I/O error: {0}")]
+    IoError(#[from] std::io::Error),
+    #[error("SSH protocol error: {0}")]
+    Protocol(#[from] russh::Error),
+}
+
+pub type SshResult<T> = Result<T, SshError>;
+
+struct Client;
+
+#[async_trait]
+impl client::Handler for Client {
+    type Error = russh::Error;
+
+    async fn check_server_key(
+        &mut self,
+        server_public_key: &key::PublicKey,
+    ) -> Result<bool, Self::Error> {
+        // Accept any server key (similar to SSH -o StrictHostKeyChecking=no)
+        // In production, you might want to verify against known_hosts
+        let _ = server_public_key; // Suppress unused warning
+        Ok(true)
+    }
+}
+
+pub struct SshClient {
+    session: client::Handle<Client>,
+}
+
+impl SshClient {
+    /// Connect to an SSH server and authenticate with password
+    pub async fn connect(host: &str, port: u32, username: &str, password: &str) -> SshResult<Self> {
+        let config = client::Config::default();
+        let sh = Client;
+
+        tracing::debug!("Connecting to {}:{} as {}", host, port, username);
+
+        let mut session = client::connect(Arc::new(config), (host, port as u16), sh)
+            .await
+            .map_err(|e| SshError::ConnectionFailed(e.to_string()))?;
+
+        let auth_result = session
+            .authenticate_password(username, password)
+            .await
+            .map_err(|e| SshError::ConnectionFailed(e.to_string()))?;
+
+        if !auth_result {
+            return Err(SshError::AuthenticationFailed);
+        }
+
+        tracing::debug!("SSH authentication successful");
+
+        Ok(Self { session })
+    }
+
+    /// Execute a command and return the output as a String
+    pub async fn execute_command(&mut self, command: &str) -> SshResult<String> {
+        tracing::debug!("Executing SSH command: {}", command);
+
+        let mut channel = self
+            .session
+            .channel_open_session()
+            .await
+            .map_err(|e| SshError::CommandFailed(e.to_string()))?;
+
+        channel
+            .exec(true, command)
+            .await
+            .map_err(|e| SshError::CommandFailed(e.to_string()))?;
+
+        let mut output = Vec::new();
+        let mut stderr_output = Vec::new();
+
+        loop {
+            let Some(msg) = channel.wait().await else {
+                break;
+            };
+
+            match msg {
+                russh::ChannelMsg::Data { ref data } => {
+                    output.extend_from_slice(data);
+                }
+                russh::ChannelMsg::ExtendedData { ref data, ext: 1 } => {
+                    // stderr
+                    stderr_output.extend_from_slice(data);
+                }
+                russh::ChannelMsg::ExitStatus { exit_status } => {
+                    tracing::debug!("Command exit status: {}", exit_status);
+                    if exit_status != 0 {
+                        let stderr_str = String::from_utf8_lossy(&stderr_output);
+                        return Err(SshError::CommandFailed(format!(
+                            "Command exited with status {}: {}",
+                            exit_status, stderr_str
+                        )));
+                    }
+                }
+                russh::ChannelMsg::Eof => {
+                    break;
+                }
+                _ => {}
+            }
+        }
+
+        channel
+            .eof()
+            .await
+            .map_err(|e| SshError::CommandFailed(e.to_string()))?;
+
+        channel
+            .close()
+            .await
+            .map_err(|e| SshError::CommandFailed(e.to_string()))?;
+
+        let output_str = String::from_utf8_lossy(&output).to_string();
+        tracing::debug!(
+            "Command output: {} bytes, {} lines",
+            output_str.len(),
+            output_str.lines().count()
+        );
+
+        Ok(output_str)
+    }
+
+    /// Close the SSH session
+    pub async fn close(self) -> SshResult<()> {
+        self.session
+            .disconnect(russh::Disconnect::ByApplication, "", "")
+            .await?;
+        Ok(())
+    }
+}

src/ssh/mod.rs

@@ -0,0 +1,3 @@
+pub mod client;
+
+pub use client::SshClient;

src/websocket_client.rs

@@ -338,6 +338,19 @@ impl AgentClient {
     }
 }
 
+/// Redact SNMP community string for logging, showing first 2 chars only
+fn redact_community(community: &str) -> String {
+    let len = community.len();
+    if len == 0 {
+        return "[redacted]".to_string();
+    }
+    if len <= 2 {
+        "**".to_string()
+    } else {
+        format!("{}***", &community[..2])
+    }
+}
+
 /// Execute an SNMP job and collect results.
 async fn execute_snmp_job(
     job: AgentJob,
@@ -346,11 +359,7 @@ async fn execute_snmp_job(
     let snmp_device = job.snmp_device.ok_or("Job missing SNMP device info")?;
 
     // Log SNMP connection parameters for debugging (mask community for security)
-    let community_masked = if snmp_device.community.len() > 4 {
+    let community_masked = redact_community(&snmp_device.community);
-        format!("{}***", &snmp_device.community[..2])
-    } else {
-        "***".to_string()
-    };
 
     tracing::info!(
         "Executing SNMP job for device {} at {}:{} (community: {}, version: {})",
@@ -470,6 +479,20 @@ async fn execute_mikrotik_job(
 ) -> Result<()> {
     use crate::mikrotik::{MikrotikClient, SecretString};
 
+    let timestamp = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)?
+        .as_secs() as i64;
+
+    // Check if this is a backup job (job_id starts with "backup:")
+    // Backup jobs use SSH instead of API because /export doesn't work via API
+    if job.job_id.starts_with("backup:") {
+        let mikrotik_device = job
+            .mikrotik_device
+            .clone()
+            .ok_or("Job missing MikroTik device info")?;
+        return execute_mikrotik_backup_via_ssh(job, mikrotik_device, result_tx, timestamp).await;
+    }
+
     let mikrotik_device = job
         .mikrotik_device
         .ok_or("Job missing MikroTik device info")?;
@@ -483,10 +506,6 @@ async fn execute_mikrotik_job(
         mikrotik_device.use_ssl
     );
 
-    let timestamp = std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)?
-        .as_secs() as i64;
-
     let password = SecretString::new(&mikrotik_device.password);
 
     // Connect and authenticate to MikroTik RouterOS API
@@ -587,6 +606,15 @@ async fn execute_mikrotik_job(
                         attr_keys
                     );
 
+                    // Log when we hit EOF during /file/read
+                    if cmd.command == "/file/read" {
+                        if let Some(data) = sentence.attributes.get("data") {
+                            if data.is_empty() {
+                                tracing::debug!("Reached end of file (empty chunk)");
+                            }
+                        }
+                    }
+
                     all_sentences.push(MikrotikSentence {
                         attributes: sentence.attributes.clone(),
                     });
@@ -631,6 +659,106 @@ async fn execute_mikrotik_job(
     Ok(())
 }
 
+/// Execute a MikroTik backup job via SSH (because /export doesn't work via API).
+async fn execute_mikrotik_backup_via_ssh(
+    job: AgentJob,
+    mikrotik_device: crate::proto::agent::MikrotikDevice,
+    result_tx: mpsc::UnboundedSender<MikrotikResult>,
+    timestamp: i64,
+) -> Result<()> {
+    use crate::ssh::SshClient;
+
+    tracing::info!(
+        "Executing backup via SSH for device {} at {}:{} (job: {})",
+        job.device_id,
+        mikrotik_device.ip,
+        mikrotik_device.ssh_port,
+        job.job_id
+    );
+
+    // Connect via SSH
+    let mut ssh_client = match SshClient::connect(
+        &mikrotik_device.ip,
+        mikrotik_device.ssh_port,
+        &mikrotik_device.username,
+        &mikrotik_device.password,
+    )
+    .await
+    {
+        Ok(client) => client,
+        Err(e) => {
+            let error_msg = format!("SSH connection failed: {}", e);
+            tracing::error!("{}", error_msg);
+            let result = MikrotikResult {
+                device_id: job.device_id,
+                job_id: job.job_id,
+                sentences: vec![],
+                error: error_msg,
+                timestamp,
+            };
+            let _ = result_tx.send(result);
+            return Err(format!("SSH connection failed: {}", e).into());
+        }
+    };
+
+    // Execute /export compact command
+    let config = match ssh_client.execute_command("/export compact").await {
+        Ok(output) => output,
+        Err(e) => {
+            let error_msg = format!("SSH command failed: {}", e);
+            tracing::error!("{}", error_msg);
+            let result = MikrotikResult {
+                device_id: job.device_id,
+                job_id: job.job_id,
+                sentences: vec![],
+                error: error_msg,
+                timestamp,
+            };
+            let _ = result_tx.send(result);
+            let _ = ssh_client.close().await;
+            return Err(format!("SSH command failed: {}", e).into());
+        }
+    };
+
+    // Close SSH connection
+    let _ = ssh_client.close().await;
+
+    tracing::info!(
+        "Backup completed: {} bytes, {} lines",
+        config.len(),
+        config.lines().count()
+    );
+
+    // Return the config as a single sentence with "config" attribute
+    let mut attributes = std::collections::HashMap::new();
+    attributes.insert("config".to_string(), config);
+
+    let job_id_for_log = job.job_id.clone();
+
+    let result = MikrotikResult {
+        device_id: job.device_id,
+        job_id: job.job_id,
+        sentences: vec![MikrotikSentence { attributes }],
+        error: String::new(),
+        timestamp,
+    };
+
+    tracing::info!(
+        "MikroTik backup job {} completed successfully",
+        result.job_id
+    );
+
+    if let Err(e) = result_tx.send(result) {
+        tracing::warn!(
+            "Failed to send MikroTik backup result for job {}: channel closed",
+            job_id_for_log
+        );
+        return Err(format!("Result channel closed: {}", e).into());
+    }
+
+    Ok(())
+}
+
 /// Convert SnmpValue to String for protobuf transmission.
 fn value_to_string(value: SnmpValue) -> String {
     match value {
@@ -908,4 +1036,30 @@ mod tests {
     }
 
     // Note: AgentClient methods require WebSocket connection and are tested via integration tests
+
+    #[test]
+    fn test_redact_community_normal() {
+        assert_eq!(redact_community("public"), "pu***");
+    }
+
+    #[test]
+    fn test_redact_community_short() {
+        assert_eq!(redact_community("ab"), "**");
+        assert_eq!(redact_community("a"), "**");
+    }
+
+    #[test]
+    fn test_redact_community_empty() {
+        assert_eq!(redact_community(""), "[redacted]");
+    }
+
+    #[test]
+    fn test_redact_community_three_chars() {
+        assert_eq!(redact_community("abc"), "ab***");
+    }
+
+    #[test]
+    fn test_redact_community_long() {
+        assert_eq!(redact_community("mysecretcommunity"), "my***");
+    }
 }