find ports in Brocade switch & configure port-channels (blind implementation)

Cargo.lock

@@ -674,6 +674,17 @@ dependencies = [
  "serde_with",
 ]
 
+[[package]]
+name = "brocade"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "harmony_types",
+ "russh",
+ "russh-keys",
+ "tokio",
+]
+
 [[package]]
 name = "brotli"
 version = "8.0.2"
@@ -2318,6 +2329,7 @@ dependencies = [
  "async-trait",
  "base64 0.22.1",
  "bollard",
+ "brocade",
  "chrono",
  "cidr",
  "convert_case",

Cargo.toml

@@ -15,7 +15,7 @@ members = [
   "harmony_inventory_agent",
   "harmony_secret_derive",
   "harmony_secret",
-  "adr/agent_discovery/mdns",
+  "adr/agent_discovery/mdns", "brocade",
 ]
 
 [workspace.package]

brocade/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "brocade"
+edition = "2024"
+version.workspace = true
+readme.workspace = true
+license.workspace = true
+
+[dependencies]
+async-trait.workspace = true
+harmony_types = { version = "0.1.0", path = "../harmony_types" }
+russh.workspace = true
+russh-keys.workspace = true
+tokio.workspace = true

brocade/src/lib.rs

@@ -0,0 +1,251 @@
+use std::{
+    fmt::{self, Display},
+    sync::Arc,
+};
+
+use async_trait::async_trait;
+use harmony_types::net::{IpAddress, MacAddress};
+use russh::client::{Handle, Handler};
+use russh_keys::key;
+use std::str::FromStr;
+
+#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone)]
+pub struct MacAddressEntry {
+    pub vlan: u16,
+    pub mac_address: MacAddress,
+    pub port_name: String,
+}
+
+pub struct BrocadeClient {
+    client: Handle<Client>,
+}
+
+impl BrocadeClient {
+    pub async fn init(ip: IpAddress, username: &str, password: &str) -> Result<Self, Error> {
+        let config = russh::client::Config::default();
+        let mut client = russh::client::connect(Arc::new(config), (ip, 22), Client {}).await?;
+
+        match client.authenticate_password(username, password).await? {
+            true => Ok(Self { client }),
+            false => Err(Error::AuthenticationError(
+                "ssh authentication failed".to_string(),
+            )),
+        }
+    }
+
+    pub async fn show_mac_address_table(&self) -> Result<Vec<MacAddressEntry>, Error> {
+        let output = self.run_command("show mac-address-table").await?;
+        let mut entries = Vec::new();
+
+        // The Brocade output usually has a header and then one entry per line.
+        // We will skip the header and parse each line.
+        // Sample line: "1234  AA:BB:CC:DD:EE:F1   GigabitEthernet1/1/1"
+        for line in output.lines().skip(1) {
+            // Skip the header row
+            let parts: Vec<&str> = line.split_whitespace().collect();
+            if parts.len() >= 3 {
+                // Assuming the format is: <VLAN> <MAC> <Port>
+                if let Ok(vlan) = u16::from_str(parts[0]) {
+                    let mac = MacAddress::try_from(parts[1].to_string());
+                    let port = parts[2].to_string();
+
+                    if let Ok(mac_address) = mac {
+                        entries.push(MacAddressEntry {
+                            vlan,
+                            mac_address,
+                            port_name: port,
+                        });
+                    }
+                }
+            }
+        }
+
+        Ok(entries)
+    }
+
+    pub async fn configure_port_channel(&self, ports: &[String]) -> Result<u8, Error> {
+        let channel_id = self.find_available_channel_id().await?;
+        let mut commands = Vec::new();
+
+        // Start configuration mode.
+        commands.push("configure terminal".to_string());
+
+        // Create the port channel interface.
+        commands.push(format!("interface Port-channel {channel_id}"));
+        commands.push("no ip address".to_string());
+        commands.push("exit".to_string());
+
+        // Configure each physical port to join the channel.
+        for port in ports {
+            commands.push(format!("interface {port}"));
+            // 'channel-group' command to add the interface to the port channel.
+            // Using 'mode active' enables LACP.
+            commands.push(format!("channel-group {channel_id} mode active"));
+            commands.push("exit".to_string());
+        }
+
+        // Save the configuration.
+        commands.push("write memory".to_string());
+
+        self.run_commands(commands).await?;
+
+        Ok(channel_id)
+    }
+
+    pub async fn find_available_channel_id(&self) -> Result<u8, Error> {
+        // FIXME: The command might vary slightly by Brocade OS version.
+        let output = self.run_command("show port-channel summary").await?;
+        let mut used_ids = Vec::new();
+
+        // Sample output line: "3   Po3(SU)   LACP       Eth  Yes      128/128  active      "
+        // We're looking for the ID, which is the first number.
+        for line in output.lines() {
+            if line.trim().starts_with(|c: char| c.is_ascii_digit()) {
+                let parts: Vec<&str> = line.split_whitespace().collect();
+                if let Ok(id) = u8::from_str(parts[0]) {
+                    used_ids.push(id);
+                }
+            }
+        }
+
+        // Sort the used IDs to find the next available number.
+        used_ids.sort();
+
+        let mut next_id = 1;
+        for &id in &used_ids {
+            if id == next_id {
+                next_id += 1;
+            } else {
+                // Found a gap, so this is our ID.
+                return Ok(next_id);
+            }
+        }
+
+        Ok(next_id)
+    }
+
+    async fn run_command(&self, command: &str) -> Result<String, Error> {
+        let mut channel = self.client.channel_open_session().await?;
+        let mut output = Vec::new();
+
+        channel.exec(true, command).await?;
+
+        loop {
+            let Some(msg) = channel.wait().await else {
+                break;
+            };
+
+            match msg {
+                russh::ChannelMsg::ExtendedData { ref data, .. }
+                | russh::ChannelMsg::Data { ref data } => {
+                    output.append(&mut data.to_vec());
+                }
+                russh::ChannelMsg::ExitStatus { exit_status } => {
+                    if exit_status != 0 {
+                        return Err(Error::CommandError(format!(
+                            "Command failed with exit status {exit_status}, output {}",
+                            String::from_utf8(output).unwrap_or_default()
+                        )));
+                    }
+                }
+                russh::ChannelMsg::Success
+                | russh::ChannelMsg::WindowAdjusted { .. }
+                | russh::ChannelMsg::Eof => {}
+                _ => {
+                    return Err(Error::UnexpectedError(format!(
+                        "Russh got unexpected msg {msg:?}"
+                    )));
+                }
+            }
+        }
+
+        channel.close().await?;
+
+        let output = String::from_utf8(output).expect("Output should be UTF-8 compatible");
+        Ok(output)
+    }
+
+    async fn run_commands(&self, commands: Vec<String>) -> Result<(), Error> {
+        let mut channel = self.client.channel_open_session().await?;
+
+        // Execute commands sequentially and check for errors immediately.
+        for command in commands {
+            let mut output = Vec::new();
+            channel.exec(true, command.as_str()).await?;
+
+            loop {
+                let Some(msg) = channel.wait().await else {
+                    break;
+                };
+
+                match msg {
+                    russh::ChannelMsg::ExtendedData { ref data, .. }
+                    | russh::ChannelMsg::Data { ref data } => {
+                        output.append(&mut data.to_vec());
+                    }
+                    russh::ChannelMsg::ExitStatus { exit_status } => {
+                        if exit_status != 0 {
+                            let output_str = String::from_utf8(output).unwrap_or_default();
+                            return Err(Error::CommandError(format!(
+                                "Command '{command}' failed with exit status {exit_status}: {output_str}",
+                            )));
+                        }
+                    }
+                    _ => {} // Ignore other messages like success or EOF for now.
+                }
+            }
+        }
+
+        channel.close().await?;
+        Ok(())
+    }
+}
+
+struct Client {}
+
+#[async_trait]
+impl Handler for Client {
+    type Error = Error;
+
+    async fn check_server_key(
+        &mut self,
+        _server_public_key: &key::PublicKey,
+    ) -> Result<bool, Self::Error> {
+        Ok(true)
+    }
+}
+
+#[derive(Debug)]
+pub enum Error {
+    NetworkError(String),
+    AuthenticationError(String),
+    ConfigurationError(String),
+    UnexpectedError(String),
+    CommandError(String),
+}
+
+impl Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Error::NetworkError(msg) => write!(f, "Network error: {msg}"),
+            Error::AuthenticationError(msg) => write!(f, "Authentication error: {msg}"),
+            Error::ConfigurationError(msg) => write!(f, "Configuration error: {msg}"),
+            Error::UnexpectedError(msg) => write!(f, "Unexpected error: {msg}"),
+            Error::CommandError(msg) => write!(f, "Command failed: {msg}"),
+        }
+    }
+}
+
+impl From<Error> for String {
+    fn from(val: Error) -> Self {
+        format!("{val}")
+    }
+}
+
+impl std::error::Error for Error {}
+
+impl From<russh::Error> for Error {
+    fn from(_value: russh::Error) -> Self {
+        Error::NetworkError("Russh client error".to_string())
+    }
+}

harmony/Cargo.toml

@@ -77,6 +77,7 @@ harmony_secret = { path = "../harmony_secret" }
 askama.workspace = true
 sqlx.workspace = true
 inquire.workspace = true
+brocade = { version = "0.1.0", path = "../brocade" }
 
 [dev-dependencies]
 pretty_assertions.workspace = true

harmony/src/domain/topology/ha_cluster.rs

@@ -1,9 +1,16 @@
 use async_trait::async_trait;
+use brocade::BrocadeClient;
 use harmony_macros::ip;
+use harmony_secret::Secret;
+use harmony_secret::SecretManager;
 use harmony_types::net::MacAddress;
 use harmony_types::net::Url;
 use log::debug;
 use log::info;
+use russh::client;
+use russh::client::Handler;
+use serde::Deserialize;
+use serde::Serialize;
 
 use crate::data::FileContent;
 use crate::executors::ExecutorError;
@@ -28,10 +35,12 @@ use super::PreparationOutcome;
 use super::Router;
 use super::Switch;
 use super::SwitchError;
+use super::SwitchPort;
 use super::TftpServer;
 
 use super::Topology;
 use super::k8s::K8sClient;
+use std::error::Error;
 use std::sync::Arc;
 
 #[derive(Debug, Clone)]
@@ -93,6 +102,39 @@ impl HAClusterTopology {
             .to_string()
     }
 
+    fn find_master_switch(&self) -> Option<LogicalHost> {
+        self.switch.first().cloned() // FIXME: Should we be smarter to find the master switch?
+    }
+
+    async fn configure_bond(&self, config: &HostNetworkConfig) -> Result<(), SwitchError> {
+        todo!()
+    }
+
+    async fn configure_port_channel(&self, config: &HostNetworkConfig) -> Result<u8, SwitchError> {
+        let auth = SecretManager::get_or_prompt::<BrocadeSwitchAuth>()
+            .await
+            .map_err(|e| SwitchError::new(format!("Failed to get credentials: {e}")))?;
+
+        let switch = self
+            .find_master_switch()
+            .ok_or(SwitchError::new("No switch found in topology".to_string()))?;
+        let client = BrocadeSwitchClient::init(switch.ip, &auth.username, &auth.password)
+            .await
+            .map_err(|e| SwitchError::new(format!("Failed to connect to switch: {e}")))?;
+
+        let switch_ports: Vec<String> = config
+            .switch_ports
+            .iter()
+            .map(|s| s.port_name.clone())
+            .collect();
+        let channel_id = client
+            .configure_port_channel(switch_ports)
+            .await
+            .map_err(|e| SwitchError::new(format!("Failed to configure switch: {e}")))?;
+
+        Ok(channel_id)
+    }
+
     pub fn autoload() -> Self {
         let dummy_infra = Arc::new(DummyInfra {});
         let dummy_host = LogicalHost {
@@ -269,19 +311,77 @@ impl HttpServer for HAClusterTopology {
 
 #[async_trait]
 impl Switch for HAClusterTopology {
-    async fn get_port_for_mac_address(&self, _mac_address: &MacAddress) -> Option<String> {
+    async fn get_port_for_mac_address(&self, mac_address: &MacAddress) -> Option<String> {
-        todo!()
+        let auth = SecretManager::get_or_prompt::<BrocadeSwitchAuth>()
+            .await
+            .unwrap();
+
+        let switch = self.find_master_switch()?;
+        let client = BrocadeSwitchClient::init(switch.ip, &auth.username, &auth.password).await;
+
+        let Ok(client) = client else {
+            return None;
+        };
+
+        client.find_port(mac_address).await
     }
 
     async fn configure_host_network(
         &self,
         _host: &PhysicalHost,
-        _config: HostNetworkConfig,
+        config: HostNetworkConfig,
     ) -> Result<(), SwitchError> {
+        let _ = self.configure_bond(&config).await;
+        let channel_id = self.configure_port_channel(&config).await;
+
         todo!()
     }
 }
 
+#[async_trait]
+trait SwitchClient {
+    async fn find_port(&self, mac_address: &MacAddress) -> Option<String>;
+    async fn configure_port_channel(&self, switch_ports: Vec<String>) -> Result<u8, SwitchError>;
+}
+
+struct BrocadeSwitchClient {
+    brocade: BrocadeClient,
+}
+
+impl BrocadeSwitchClient {
+    async fn init(ip: IpAddress, username: &str, password: &str) -> Result<Self, brocade::Error> {
+        let brocade = BrocadeClient::init(ip, username, password).await?;
+        Ok(Self { brocade })
+    }
+}
+
+#[async_trait]
+impl SwitchClient for BrocadeSwitchClient {
+    async fn find_port(&self, mac_address: &MacAddress) -> Option<String> {
+        let Ok(table) = self.brocade.show_mac_address_table().await else {
+            return None;
+        };
+
+        table
+            .iter()
+            .find(|entry| entry.mac_address == *mac_address)
+            .map(|entry| entry.port_name.clone())
+    }
+
+    async fn configure_port_channel(&self, switch_ports: Vec<String>) -> Result<u8, SwitchError> {
+        self.brocade
+            .configure_port_channel(&switch_ports)
+            .await
+            .map_err(|e| SwitchError::new(format!("Failed to configure port channel: {e}")))
+    }
+}
+
+#[derive(Secret, Serialize, Deserialize, Debug)]
+struct BrocadeSwitchAuth {
+    username: String,
+    password: String,
+}
+
 #[derive(Debug)]
 pub struct DummyInfra;
 

harmony_types/src/net.rs

@@ -1,6 +1,6 @@
 use serde::{Deserialize, Serialize};
 
-#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize, PartialOrd, Ord)]
 pub struct MacAddress(pub [u8; 6]);
 
 impl MacAddress {