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.

changes/429.security

@@ -0,0 +1 @@
+Stopped the source server password appearing in the logs during an FTP, HTTP or HTTPS file copy to Cisco IOS devices.

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

@@ -6,6 +6,7 @@
 import warnings
 
 from netmiko import ConnectHandler, FileTransfer
+from netmiko.base_connection import SecretsFilter
 from netmiko.exceptions import ReadTimeout
 
 from pyntc import log
@@ -36,6 +37,65 @@
 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"}
+# Key the copy URL password is filed under in netmiko's no_log mapping.
+FILE_COPY_NO_LOG_KEY = "file_copy_token"
+
+
+class HideUrlCredentials:
+    """Hide a copy URL password from pyntc and netmiko log records.
+
+    FTP, HTTP and HTTPS carry the password in the copy command itself, so it reaches
+    netmiko's channel logging, netmiko's session log, and the device output this
+    module logs. Netmiko's `SecretsFilter` rewrites the message of each record logged
+    through the logger it is attached to, so one filter goes on the pyntc logger, and
+    the token joins the mapping the connection's own filter and session log share.
+    """
+
+    def __init__(self, native, token):
+        """Capture the registries the token is added to and removed from.
+
+        Args:
+            native: The netmiko connection whose no_log mapping the token joins.
+            token (str): The password sent in the copy URL. A falsy value is a no-op.
+        """
+        self.token = token
+        self.netmiko_no_log = native._secrets_filter.no_log  # pylint: disable=protected-access
+        self.pyntc_log = log.get_log()
+        self.pyntc_filter = None
+
+    def __enter__(self):
+        """Add a `SecretsFilter` to the pyntc logger and the token to netmiko's mapping.
+
+        The pyntc filter covers the messages logged here. The netmiko mapping is the one
+        its own `SecretsFilter` and its `SessionLog` read from, so writing the token
+        there covers the copy command as netmiko sends it.
+
+        Returns:
+            HideUrlCredentials: This instance.
+        """
+        if not self.token:
+            return self
+        self.pyntc_filter = SecretsFilter(no_log={FILE_COPY_NO_LOG_KEY: self.token})
+        self.netmiko_no_log[FILE_COPY_NO_LOG_KEY] = self.token
+        self.pyntc_log.addFilter(self.pyntc_filter)
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        """Unregister the token.
+
+        Returns:
+            bool: False, so an exception is never suppressed.
+        """
+        if self.pyntc_filter is None:
+            return False
+        self.netmiko_no_log.pop(FILE_COPY_NO_LOG_KEY, None)
+        self.pyntc_log.removeFilter(self.pyntc_filter)
+        return False
 
 
 @fix_docs
@@ -795,7 +855,46 @@ 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 put in an exception message.
+
+        A logging filter cannot reach an exception message, so the masking happens here.
+        """
+        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,too-many-locals
+        self, src: FileCopyModel, dest=None, file_system=None, **kwargs
+    ):
         """Copy a file to a remote device.
 
         Args:
@@ -824,50 +923,72 @@ 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"}:
+            credentials_in_url = bool(src.username) and src.scheme in IOS_URL_CREDENTIAL_SCHEMES
+            if credentials_in_url:
+                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
-            # there may be a % warning that does not indicate a failure so we will use send_command directly.
-            output = self.native.send_command(command, expect_string=expect_regex, read_timeout=src.timeout)
+            with HideUrlCredentials(self.native, src.token if credentials_in_url else None):
+                # _send_command raises on "% ", and a % warning during a copy is not a failure.
+                output = self.native.send_command(command, expect_string=expect_regex, read_timeout=src.timeout)
 
-            while current_prompt not in output:
-                # Check for success message in output to break loop and avoid waiting for next prompt
-                if re.search(r"Copy complete|bytes copied in|File transfer successful", output, re.IGNORECASE):
-                    log.info(
-                        "Host %s: File %s transferred successfully with output: %s", self.host, src.file_name, output
-                    )
-                    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
-                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
+                while current_prompt not in output:
+                    # Break on the success marker rather than waiting for the next prompt.
+                    if re.search(r"Copy complete|bytes copied in|File transfer successful", output, re.IGNORECASE):
+                        # SecretsFilter rewrites a record's message and never its args, so output goes in the message.
+                        message = (
+                            f"Host {self.host}: File {src.file_name} transferred successfully with output: {output}"
+                        )
+                        log.info(message)
+                        break
+                    # Raise on an error marker so the failure reports what the device said.
+                    if re.search(r"(Error|Invalid|Failed|Aborted|denied)", output, re.IGNORECASE):
+                        message = f"Host {self.host}: File transfer error for {src.file_name}: {output}"
+                        log.error(message)
+                        raise FileTransferError(
+                            f"Error detected in copy command output: {self._mask_token(output, src)}"
+                        )
+                    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
+                            )
+                            # output has been replaced, so re-test it instead of the remaining prompts.
+                            break
+                    else:
+                        # Nothing matched, so output never changes and the loop would spin forever.
+                        message = (
+                            f"Host {self.host}: Unexpected output during file transfer of {src.file_name}: {output}"
+                        )
+                        log.error(message)
+                        raise FileTransferError(
+                            f"Unexpected output during file transfer: {self._mask_token(output, src)}"
                         )
-                        break  # Exit the for loop and check the new output for the next prompt
 
             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)