Merge pull request #3691 from FestplattenSchnitzel/aruba_os_file_transfer

Aruba OS: Add file transfer support

Author: ktbyers

PLATFORMS.md

@@ -311,6 +311,7 @@
 
 ###### Supported Secure Copy device_type values
 
+- aruba_os
 - arista_eos
 - ciena_saos
 - cisco_asa

netmiko/aruba/__init__.py

@@ -1,4 +1,4 @@
 from netmiko.aruba.aruba_aoscx import ArubaCxSSH
-from netmiko.aruba.aruba_os import ArubaOsSSH
+from netmiko.aruba.aruba_os import ArubaOsSSH, ArubaOsFileTransfer
 
-__all__ = ["ArubaOsSSH", "ArubaCxSSH"]
+__all__ = ["ArubaOsSSH", "ArubaCxSSH", "ArubaOsFileTransfer"]

netmiko/aruba/aruba_os.py

@@ -5,8 +5,11 @@
 
 """
 
-from typing import Any
+import os
+from typing import Any, Optional
+
 from netmiko.cisco_base_connection import CiscoSSHConnection
+from netmiko.scp_handler import BaseFileTransfer
 
 
 class ArubaOsSSH(CiscoSSHConnection):
@@ -52,3 +55,152 @@ def config_mode(
         return super().config_mode(
             config_command=config_command, pattern=pattern, re_flags=re_flags
         )
+
+
+class ArubaOsFileTransfer(BaseFileTransfer):
+    """Aruba OS SCP File Transfer driver"""
+
+    def __init__(
+        self,
+        file_system: Optional[str] = "/mm/mynode",
+        hash_supported: bool = False,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(
+            file_system=file_system, hash_supported=hash_supported, **kwargs
+        )
+
+    def file_md5(self, file_name: str, add_newline: bool = False) -> str:
+        msg = "Aruba OS does not support an MD5-hash operation."
+        raise AttributeError(msg)
+
+    @staticmethod
+    def process_md5(md5_output: str, pattern: str = "") -> str:
+        msg = "Aruba OS does not support an MD5-hash operation."
+        raise AttributeError(msg)
+
+    def compare_md5(self) -> bool:
+        msg = "Aruba OS does not support an MD5-hash operation."
+        raise AttributeError(msg)
+
+    def remote_md5(self, base_cmd: str = "", remote_file: Optional[str] = None) -> str:
+        msg = "Aruba OS does not support an MD5-hash operation."
+        raise AttributeError(msg)
+
+    def check_file_exists(self, remote_cmd: str = "") -> bool:
+        """Check if the dest_file already exists on the file system (return boolean)."""
+        if self.direction == "put":
+            if not remote_cmd:
+                remote_cmd = f"dir search {self.dest_file.rpartition('/')[-1]}"
+            remote_out = self.ssh_ctl_chan._send_command_str(remote_cmd)
+
+            if "Cannot get directory information" in remote_out:
+                return False
+
+            return self.dest_file in [
+                split_line[-1]
+                for line in remote_out.splitlines()
+                if (split_line := line.split())
+            ]
+        elif self.direction == "get":
+            return os.path.exists(self.dest_file)
+        else:
+            raise ValueError("Unexpected value for self.direction")
+
+    def remote_file_size(
+        self, remote_cmd: str = "", remote_file: Optional[str] = None
+    ) -> int:
+        """Get the file size of the remote file."""
+        if remote_file is None:
+            if self.direction == "put":
+                remote_file = self.dest_file
+            elif self.direction == "get":
+                remote_file = self.source_file
+
+        assert isinstance(remote_file, str)
+        remote_file_search = remote_file.rpartition("/")[-1]
+
+        if not remote_cmd:
+            remote_cmd = f"dir search {remote_file_search}"
+        remote_out = self.ssh_ctl_chan._send_command_str(remote_cmd)
+
+        if "Cannot get directory information" in remote_out:
+            msg = "Unable to find file on remote system"
+            raise IOError(msg)
+
+        file_size = [
+            split_line[-5]
+            for line in remote_out.splitlines()
+            if (split_line := line.split()) and split_line[-1] == remote_file_search
+        ]
+
+        if len(file_size) != 1:
+            msg = (
+                "Unable to parse remote file size, found file count is not equal to one"
+            )
+            raise IOError(msg)
+
+        try:
+            return int(file_size[0])
+        except ValueError as ve:
+            msg = "Unable to parse remote file size, wrong field in use or malformed command output"
+            raise IOError(msg) from ve
+
+    def verify_file(self) -> bool:
+        """Verify the file has been transferred correctly based on filesize."""
+        if self.direction == "put":
+            return os.stat(self.source_file).st_size == self.remote_file_size(
+                remote_file=self.dest_file
+            )
+        elif self.direction == "get":
+            return (
+                self.remote_file_size(remote_file=self.source_file)
+                == os.stat(self.dest_file).st_size
+            )
+        else:
+            raise ValueError("Unexpected value of self.direction")
+
+    def remote_space_available(self, search_pattern: str = "") -> int:
+        """Return space available on remote device."""
+        remote_cmd = "show storage"
+        remote_output = self.ssh_ctl_chan._send_command_str(remote_cmd).strip()
+
+        # df -h ouput
+        available_sizes = [
+            split_line[-3]
+            for line in remote_output.splitlines()
+            if (split_line := line.split()) and split_line[-1] == "/flash"
+        ]
+
+        if not available_sizes:
+            msg = "Could not determine remote space available."
+            raise ValueError(msg)
+
+        space_available = 0
+
+        for available_size in available_sizes:
+            size_names = ["B", "K", "M", "G", "T", "P", "E", "Z", "Y"]
+
+            size_str, suffix = available_size[:-1], available_size[-1]
+
+            if suffix not in size_names:
+                msg = "Could not determine remote space available."
+                raise ValueError(msg)
+
+            try:
+                size = float(size_str)
+            except ValueError as ve:
+                msg = "Could not determine remote space available."
+                raise ValueError(msg) from ve
+
+            space_available += size * 1024 ** size_names.index(suffix)
+
+        return int(space_available)
+
+    def enable_scp(self, cmd: str = "service scp") -> None:
+        """Enable SCP on remote device."""
+        super().enable_scp(cmd)
+
+    def disable_scp(self, cmd: str = "no service scp") -> None:
+        """Disable SCP on remote device."""
+        super().disable_scp(cmd)

netmiko/ssh_dispatcher.py

@@ -16,7 +16,7 @@
 from netmiko.arista import AristaFileTransfer
 from netmiko.arris import ArrisCERSSH
 from netmiko.apresia import ApresiaAeosSSH, ApresiaAeosTelnet
-from netmiko.aruba import ArubaOsSSH, ArubaCxSSH
+from netmiko.aruba import ArubaOsSSH, ArubaCxSSH, ArubaOsFileTransfer
 from netmiko.asterfusion import AsterfusionAsterNOSSSH
 from netmiko.audiocode import (
     Audiocode72SSH,
@@ -337,6 +337,7 @@
 }
 
 FILE_TRANSFER_MAP = {
+    "aruba_os": ArubaOsFileTransfer,
     "arista_eos": AristaFileTransfer,
     "ciena_saos": CienaSaosFileTransfer,
     "cisco_asa": CiscoAsaFileTransfer,