Add remote file copy for cisco nxos devices

Author: mattmiller87

pyntc/devices/nxos_device.py

@@ -4,6 +4,10 @@
 import re
 import time
 
+from urllib.parse import urlparse
+from netmiko import ConnectHandler
+from pyntc.utils.models import FileCopyModel
+
 from pynxos.device import Device as NXOSNative
 from pynxos.errors import CLIError
 from pynxos.features.file_copy import FileTransferError as NXOSFileTransferError
@@ -300,6 +304,222 @@ def file_copy_remote_exists(self, src, dest=None, file_system="bootflash:"):
         )
         return self.native.file_copy_remote_exists(src, dest, file_system=file_system)
 
+    ###
+    # MM Begin
+    ###
+    def check_file_exists(self, filename, file_system=None):
+        """Check if a remote file exists by filename.
+
+        Args:
+            filename (str): The name of the file to check for on the remote device.
+            file_system (str): Supported only for Arista. The file system for the
+                remote file. If no file_system is provided, then the `get_file_system`
+                method is used to determine the correct file system to use.
+
+        Returns:
+            (bool): True if the remote file exists, False if it doesn't.
+
+        Raises:
+            CommandError: If there is an error in executing the command to check if the file exists.
+        """
+        exists = False
+
+        file_system = file_system or self._get_file_system()
+        command = f"dir {file_system}/{filename}"
+        result = self.native_ssh.send_command(command, read_timeout=30)
+
+        log.debug(
+            "Host %s: Checking if file %s exists on remote with command '%s' and result: %s",
+            self.host,
+            filename,
+            command,
+            result,
+        )
+
+        # Check for error patterns
+        if re.search(r"% Error listing directory|No such file|No files found|Path does not exist", result):
+            log.debug("Host %s: File %s does not exist on remote.", self.host, filename)
+            exists = False
+        elif re.search(rf"Directory of .*{filename}", result):
+            log.debug("Host %s: File %s exists on remote.", self.host, filename)
+            exists = True
+        else:
+            raise CommandError(command, f"Unable to determine if file {filename} exists on remote: {result}")
+
+        return exists
+
+    def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs):
+        # Validate hashing algorithm
+        if hashing_algorithm not in {"md5", "sha512"}:
+            log.error(
+                "Host %s: Unsupported hashing algorithm %s. Supported algorithms are: md5, sha512",
+                self.host,
+                hashing_algorithm,
+            )
+            raise ValueError(f"hashing_algorithm must be either 'md5' or 'sha512', got '{hashing_algorithm}'")
+
+        # Determine file system
+        if file_system is None:
+            file_system = self._get_file_system()
+        
+        # Normalize file_system
+        if not file_system.startswith("/") and not file_system.endswith(":"):
+            file_system = f"{file_system}:"
+
+        # Use NXOS verify command to get the checksum
+        # Example: show file bootflash:nautobot.png sha512sum
+        command = f"show file {file_system}/{filename} {hashing_algorithm}sum"
+
+        try:
+            result = self.native_ssh.send_command(command, read_timeout=30)
+            log.debug(
+                "Host %s: Getting remote checksum for file %s with command '%s' and result: %s",
+                self.host,
+                filename,
+                command,
+                result,
+            )
+
+            # Parse the output to extract the checksum
+            match = re.search(r"=\s*([a-fA-F0-9]+)", result)
+            if match:
+                remote_checksum = match.group(1).lower()
+                log.debug("Host %s: Remote checksum for %s: %s", self.host, filename, remote_checksum)
+                return remote_checksum
+
+            log.error("Host %s: Could not parse checksum from output: %s", self.host, result)
+            raise CommandError(command, f"Could not parse checksum from output: {result}")
+
+        except Exception as e:
+            log.error("Host %s: Error getting remote checksum: %s", self.host, str(e))
+            raise CommandError(command, f"Error getting remote checksum: {str(e)}")
+
+    def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kwargs):
+        """Copy a file from remote source to device.
+
+        Args:
+            src (FileCopyModel): The source file model with transfer parameters.
+            dest (str): Destination filename (defaults to src.file_name).
+            file_system (str): Device filesystem (auto-detected if not provided).
+            **kwargs (Any): Passible parameters such as file_system.
+
+        Raises:
+            TypeError: If src is not a FileCopyModel.
+            FileTransferError: If transfer or verification fails.
+            FileSystemNotFoundError: If filesystem cannot be determined.
+        """
+        # Validate input
+        if not isinstance(src, FileCopyModel):
+            raise TypeError("src must be an instance of FileCopyModel")
+
+        # Determine file system
+        if file_system is None:
+            file_system = self._get_file_system()
+
+        # Determine destination
+        if dest is None:
+            dest = src.file_name
+
+        # Validate scheme
+        supported_schemes = ["http", "https", "scp", "ftp", "sftp", "tftp"]
+        if src.scheme not in supported_schemes:
+            raise ValueError(f"Unsupported scheme: {src.scheme}")
+
+        log.debug(
+            "Host %s: Verifying file %s exists before attempting a copy",
+            self.host,
+            dest,
+        )
+        if not self.verify_file(src.checksum, dest, hashing_algorithm=src.hashing_algorithm, file_system=file_system):
+            current_prompt = self.native.find_prompt()
+            prompt_answers = {
+                r"Password": src.token or "",
+                r"Source username": src.username or "",
+                r"yes/no|Are you sure you want to continue connecting": "yes",
+                r"(confirm|Address or name of remote host|Source filename|Destination filename)": "",
+            }
+            keys = list(prompt_answers.keys()) + [re.escape(current_prompt)]
+            expect_regex = f"({'|'.join(keys)})"
+        
+        # Construct copy command
+        cmd = f"copy {src.clean_url} {file_system}"
+        if src.vrf:
+            cmd += f" vrf {src.vrf}"
+
+        log.debug("Host %s: Starting remote file copy for %s to %s/%s", self.host, src.file_name, file_system, dest)
+        try:
+            result = self.native_netmiko.send_command_timing(cmd, read_timeout=src.timeout)
+        except Exception as e:
+            log.error(
+                "Host %s: File transfer error for file %s: %s",
+                self.host,
+                dest,
+                str(e),
+            )
+            raise FileTransferError(str(e))
+
+        log.debug(
+            "Host %s: Copy command output: %s",
+            self.host,
+            result,
+        )
+
+        # Check for success indicators
+        if "copy complete" not in result.lower():
+            log.error(
+                "Host %s: File transfer did not complete successfully for file %s",
+                self.host,
+                dest,
+            )
+            raise FileTransferError("Copy command did not complete successfully")
+
+        # Verify file after transfer
+        if not self.verify_file(src.checksum, dest, hashing_algorithm=src.hashing_algorithm, file_system=file_system):
+            log.error(
+                "Host %s: File verification failed after transfer for file %s",
+                self.host,
+                dest,
+            )
+            raise FileTransferError("File verification failed after transfer")
+
+        log.info(
+            "Host %s: File %s transferred successfully.",
+            self.host,
+            dest,
+        )
+
+    def verify_file(self, checksum, filename, hashing_algorithm="md5", **kwargs):
+        """Verify a file on the device by comparing checksums.
+
+        Args:
+            checksum (str): The expected checksum of the file.
+            filename (str): The name of the file on the device.
+            hashing_algorithm (str): The hashing algorithm to use (default: "md5").
+            **kwargs (Any):  Passible parameters such as file_system.
+
+        Returns:
+            (bool): True if the file is verified successfully, False otherwise.
+        """
+        exists = self.check_file_exists(filename, **kwargs)
+        device_checksum = (
+            self.get_remote_checksum(filename, hashing_algorithm=hashing_algorithm, **kwargs) if exists else None
+        )
+        if checksum == device_checksum:
+            log.debug("Host %s: Checksum verification successful for file %s", self.host, filename)
+            return True
+
+        log.debug(
+            "Host %s: Checksum verification failed for file %s - Expected: %s, Actual: %s",
+            self.host,
+            filename,
+            checksum,
+            device_checksum,
+        )
+        return False
+    ###
+    # MM End
+    ###
+
     def install_os(self, image_name, **vendor_specifics):
         """Upgrade device with provided image.