Fix IOS remote file copy credentials and report device copy errors

changes/429.fixed

@@ -0,0 +1 @@
+Fixed FTP, HTTP and HTTPS file transfers to Cisco IOS devices failing to authenticate.

changes/429.fixed.1

@@ -0,0 +1 @@
+Fixed remote file copy failures on Cisco IOS, NX-OS, ASA and IOS-XR reporting a generic message instead of the error the device returned.

changes/429.fixed.2

@@ -0,0 +1 @@
+Fixed remote file copy hanging on Cisco IOS and NX-OS when a device returned output the driver did not recognize.

pyntc/devices/asa_device.py

@@ -1039,6 +1039,11 @@ def reboot_standby(self, acceptable_states: Optional[Iterable[str]] = None, time
 
         log.debug("Host %s: reboot standby with timeout %s.", self.host, timeout)
 
+    @staticmethod
+    def _mask_token(output: str, src: FileCopyModel) -> str:
+        """Replace the token in device output, so it is safe to log or raise."""
+        return output.replace(src.token, "*****") if src.token else output
+
     def remote_file_copy(self, src: FileCopyModel = None, dest=None, **kwargs: Any):
         """Copy a file from a remote server to the device.
 
@@ -1104,8 +1109,9 @@ def remote_file_copy(self, src: FileCopyModel = None, dest=None, **kwargs: Any):
                     break
 
                 if re.search(r"(Error|Invalid|Failed|Aborted|denied)", output, re.IGNORECASE):
-                    log.error("Host %s: File transfer error for %s: %s", self.host, src.file_name, output)
-                    raise FileTransferError
+                    masked_output = self._mask_token(output, src)
+                    log.error("Host %s: File transfer error for %s: %s", self.host, src.file_name, masked_output)
+                    raise FileTransferError(f"Error detected in copy command output: {masked_output}")
 
                 for prompt, answer in prompt_answers.items():
                     if re.search(prompt, output, re.IGNORECASE):
@@ -1117,16 +1123,22 @@ def remote_file_copy(self, src: FileCopyModel = None, dest=None, **kwargs: Any):
                         )
                         break
                 else:
+                    masked_output = self._mask_token(output, src)
                     log.error(
-                        "Host %s: Unexpected output during file transfer of %s: %s", self.host, src.file_name, output
+                        "Host %s: Unexpected output during file transfer of %s: %s",
+                        self.host,
+                        src.file_name,
+                        masked_output,
                     )
-                    raise FileTransferError
+                    raise FileTransferError(f"Unexpected output during file transfer: {masked_output}")
 
             if not self.verify_file(
                 src.checksum, dest, hashing_algorithm=src.hashing_algorithm, file_system=file_system
             ):
                 log.error("Host %s: File %s could not be verified after transfer.", self.host, src.file_name)
-                raise FileTransferError
+                raise FileTransferError(
+                    f"Could not validate {src.file_name} existed and matched the expected checksum after transfer."
+                )
 
     @property
     def redundancy_mode(self):

pyntc/devices/ios_device.py

@@ -36,6 +36,11 @@
 RE_REDUNDANCY_STATE = re.compile(r"^\s*Current\s+Software\s+state\s*=\s*(.+?)\s*$", re.M)
 SHOW_DIR_RETRY_COUNT = 5
 INSTALL_MODE_FILE_NAME = "packages.conf"
+# Schemes where IOS reads the credentials out of the URL and never prompts for them.
+# Sending a bare URL for one of these makes the device attempt an anonymous login.
+IOS_URL_CREDENTIAL_SCHEMES = {"ftp", "http", "https"}
+# Schemes whose copy command rejects a trailing "vrf" keyword.
+IOS_NO_VRF_SCHEMES = {"http", "https"}
 
 
 @fix_docs
@@ -795,7 +800,43 @@ def file_copy(self, src, dest=None, file_system=None):
                 )
                 raise FileTransferError
 
-    def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kwargs):
+    @staticmethod
+    def _netloc(src: FileCopyModel) -> str:
+        """Return host:port or just host from a FileCopyModel."""
+        return f"{src.hostname}:{src.port}" if src.port else src.hostname
+
+    @staticmethod
+    def _source_path(src: FileCopyModel, dest: str) -> str:
+        """Return the file path from the URL, falling back to dest if empty."""
+        return src.path if src.path and src.path != "/" else f"/{dest}"
+
+    @staticmethod
+    def _mask_token(output: str, src: FileCopyModel) -> str:
+        """Replace the token in device output, so it is safe to log or raise."""
+        return output.replace(src.token, "*****") if src.token else output
+
+    def _build_url_copy_command_simple(self, src: FileCopyModel, file_system: str, dest: str) -> str:
+        """Build the copy command for transfers where IOS prompts for the credentials it needs.
+
+        SCP and SFTP prompt for the source username and the password, and
+        `remote_file_copy` answers both from the model.
+        """
+        return f"copy {src.clean_url} {file_system}{dest}"
+
+    def _build_url_copy_command_with_creds(self, src: FileCopyModel, file_system: str, dest: str) -> str:
+        """Build the copy command for transfers where IOS reads the credentials from the URL.
+
+        FTP, HTTP and HTTPS never prompt. A URL without credentials makes the device
+        attempt an anonymous login, which the server rejects.
+        """
+        netloc = self._netloc(src)
+        path = self._source_path(src, dest)
+        credentials = f"{src.username}:{src.token}" if src.token else src.username
+        return f"copy {src.scheme}://{credentials}@{netloc}{path} {file_system}{dest}"
+
+    def remote_file_copy(  # noqa: R0912 pylint: disable=too-many-branches
+        self, src: FileCopyModel, dest=None, file_system=None, **kwargs
+    ):
         """Copy a file to a remote device.
 
         Args:
@@ -824,16 +865,19 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kw
 
             # Define prompt mapping for expected prompts during file copy
             prompt_answers = {
-                r"Password": src.token,
-                r"Source username": src.username,
+                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)": "",  # Press Enter
             }
             keys = list(prompt_answers.keys()) + [re.escape(current_prompt)]
             expect_regex = f"({'|'.join(keys)})"
 
-            command = f"copy {src.clean_url} {file_system}{dest}"
-            if src.vrf and src.scheme not in {"http", "https"}:
+            if src.username and src.scheme in IOS_URL_CREDENTIAL_SCHEMES:
+                command = self._build_url_copy_command_with_creds(src, file_system, dest)
+            else:
+                command = self._build_url_copy_command_simple(src, file_system, dest)
+            if src.vrf and src.scheme not in IOS_NO_VRF_SCHEMES:
                 command = f"{command} vrf {src.vrf}"
 
             # _send_command currently checks for % and raises an error, but during the file copy
@@ -849,25 +893,40 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kw
                     break
                 # Check for errors explicitly to avoid infinite loops on failure
                 if re.search(r"(Error|Invalid|Failed|Aborted|denied)", output, re.IGNORECASE):
-                    log.error("Host %s: File transfer error %s", self.host, FileTransferError.default_message)
-                    raise FileTransferError
+                    masked_output = self._mask_token(output, src)
+                    log.error("Host %s: File transfer error for %s: %s", self.host, src.file_name, masked_output)
+                    raise FileTransferError(f"Error detected in copy command output: {masked_output}")
                 for prompt, answer in prompt_answers.items():
                     if re.search(prompt, output, re.IGNORECASE):
                         is_password = "Password" in prompt
                         output = self.native.send_command(
                             answer, expect_string=expect_regex, read_timeout=src.timeout, cmd_verify=not is_password
                         )
                         break  # Exit the for loop and check the new output for the next prompt
+                else:
+                    # No prompt matched and no marker was found. Without this the loop
+                    # never reassigns output and spins on the same string forever.
+                    masked_output = self._mask_token(output, src)
+                    log.error(
+                        "Host %s: Unexpected output during file transfer of %s: %s",
+                        self.host,
+                        src.file_name,
+                        masked_output,
+                    )
+                    raise FileTransferError(f"Unexpected output during file transfer: {masked_output}")
 
             if not self.verify_file(
                 src.checksum, dest, hashing_algorithm=src.hashing_algorithm, file_system=file_system
             ):
                 log.error(
-                    "Host %s: Attempted remote file copy, but could not validate file existed after transfer %s",
+                    "Host %s: Attempted remote file copy, but could not validate %s%s after transfer.",
                     self.host,
-                    FileTransferError.default_message,
+                    file_system,
+                    dest,
+                )
+                raise FileTransferError(
+                    f"Could not validate {file_system}{dest} existed and matched the expected checksum after transfer."
                 )
-                raise FileTransferError
 
     # TODO: Make this an internal method since exposing file_copy should be sufficient
     def file_copy_remote_exists(self, src, dest=None, file_system=None):

pyntc/devices/iosxr_device.py

@@ -503,6 +503,11 @@ def verify_file(self, checksum, filename, hashing_algorithm="md5", file_system=N
             checksum, filename, hashing_algorithm, file_system=file_system, read_timeout=read_timeout
         )
 
+    @staticmethod
+    def _mask_token(output: str, src: FileCopyModel) -> str:
+        """Replace the token in device output, so it is safe to log or raise."""
+        return output.replace(src.token, "*****") if src.token else output
+
     def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kwargs):
         """Copy a file from a remote URL onto the device filesystem.
 
@@ -568,8 +573,9 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kw
                 output,
                 flags=re.IGNORECASE,
             ):
-                log.error("Host %s: File transfer error for %s: %s", self.host, dest, output)
-                raise FileTransferError
+                masked_output = self._mask_token(output, src)
+                log.error("Host %s: File transfer error for %s: %s", self.host, dest, masked_output)
+                raise FileTransferError(f"Error detected in copy command output: {masked_output}")
             for prompt, answer in prompt_answers.items():
                 if re.search(prompt, output, re.IGNORECASE):
                     is_password = "password" in output.lower()
@@ -583,12 +589,13 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kw
 
         if not self.verify_file(src.checksum, dest, hashing_algorithm=src.hashing_algorithm, file_system=file_system):
             log.error(
-                "Host %s: File %s could not be verified after transfer (missing or checksum mismatch). %s",
+                "Host %s: File %s could not be verified after transfer (missing or checksum mismatch).",
                 self.host,
                 dest,
-                FileTransferError.default_message,
             )
-            raise FileTransferError
+            raise FileTransferError(
+                f"Could not validate {file_system}/{dest} existed and matched the expected checksum after transfer."
+            )
 
         log.info("Host %s: File %s copied to %s and checksum verified.", self.host, dest, file_system)
 

pyntc/devices/nxos_device.py

@@ -419,6 +419,11 @@ def _source_path(src: FileCopyModel, dest: str) -> str:
         """Return the file path from the URL, falling back to dest if empty."""
         return src.path if src.path and src.path != "/" else f"/{dest}"
 
+    @staticmethod
+    def _mask_token(output: str, src: FileCopyModel) -> str:
+        """Replace the token in device output, so it is safe to log or raise."""
+        return output.replace(src.token, "*****") if src.token else output
+
     def _build_url_copy_command_simple(self, src, file_system, dest):
         """Build copy command for simple URL-based transfers (TFTP, HTTP, HTTPS without credentials)."""
         netloc = self._netloc(src)
@@ -538,7 +543,9 @@ def get_remote_checksum(self, filename, hashing_algorithm="md5", **kwargs):
             raise CommandError(command, f"Could not parse checksum from device output: {result}")
         return match.group(1)
 
-    def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kwargs):  # noqa: R0912 pylint: disable=too-many-branches
+    def remote_file_copy(  # noqa: R0912 pylint: disable=too-many-branches,too-many-locals
+        self, src: FileCopyModel, dest=None, file_system=None, **kwargs
+    ):
         """Copy a file from remote source to device.  Skips if file already exists and is verified on remote device.
 
         Args:
@@ -614,15 +621,27 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kw
                     break
                 # Check for errors explicitly to avoid infinite loops on failure
                 if re.search(r"(Error|Invalid|Failed|Aborted|denied)", output, re.IGNORECASE):
-                    log.error("Host %s: File transfer error %s", self.host, FileTransferError.default_message)
-                    raise FileTransferError
+                    masked_output = self._mask_token(output, src)
+                    log.error("Host %s: File transfer error for %s: %s", self.host, src.file_name, masked_output)
+                    raise FileTransferError(f"Error detected in copy command output: {masked_output}")
                 for prompt, answer in prompt_answers.items():
                     if re.search(prompt, output, re.IGNORECASE):
                         is_password = "Password" in prompt
                         output = self.native_ssh.send_command(
                             answer, expect_string=expect_regex, read_timeout=timeout, cmd_verify=not is_password
                         )
                         break  # Exit the for loop and check the new output for the next prompt
+                else:
+                    # No prompt matched and no marker was found. Without this the loop
+                    # never reassigns output and spins on the same string forever.
+                    masked_output = self._mask_token(output, src)
+                    log.error(
+                        "Host %s: Unexpected output during file transfer of %s: %s",
+                        self.host,
+                        src.file_name,
+                        masked_output,
+                    )
+                    raise FileTransferError(f"Unexpected output during file transfer: {masked_output}")
 
             # Verify file after transfer
             if not self.verify_file(