From 0d45c521d2ff12cf49d00bf1fc485648a66792f4 Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Tue, 15 Sep 2026 15:33:38 -0400 Subject: [PATCH 1/2] Fix IOS remote file copy dropping URL credentials --- changes/429.fixed | 1 + pyntc/devices/ios_device.py | 49 ++++++- tests/integration/test_ios_device.py | 163 +++++++++++++++++++++ tests/unit/test_devices/test_ios_device.py | 132 +++++++++++++++++ 4 files changed, 340 insertions(+), 5 deletions(-) create mode 100644 changes/429.fixed create mode 100644 tests/integration/test_ios_device.py diff --git a/changes/429.fixed b/changes/429.fixed new file mode 100644 index 00000000..955324e6 --- /dev/null +++ b/changes/429.fixed @@ -0,0 +1 @@ +Fixed FTP, HTTP and HTTPS file transfers to Cisco IOS devices failing to authenticate. diff --git a/pyntc/devices/ios_device.py b/pyntc/devices/ios_device.py index 362de9f3..a257af5c 100644 --- a/pyntc/devices/ios_device.py +++ b/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,38 @@ 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}" + + 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 +860,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 diff --git a/tests/integration/test_ios_device.py b/tests/integration/test_ios_device.py new file mode 100644 index 00000000..884bf02d --- /dev/null +++ b/tests/integration/test_ios_device.py @@ -0,0 +1,163 @@ +"""Integration tests for IOSDevice.remote_file_copy. + +These tests connect to an actual Cisco IOS device in the lab and are run manually. +They are NOT part of the CI unit test suite. + +Usage (from project root): + export IOS_HOST= + export IOS_USER= + export IOS_PASS= + export FTP_URL=ftp://:@/ + export TFTP_URL=tftp:/// + export SCP_URL=scp://:@:2022/ + export HTTP_URL=http://:@:8081/ + export HTTPS_URL=https://:@:8443/ + export SFTP_URL=sftp://:@:2022/ + export FILE_CHECKSUM_MD5= + export FILE_SIZE= + export FILE_SIZE_UNIT=bytes # optional; defaults to "bytes" + # export IOS_VRF=Mgmt-vrf # optional; applied to every copy test + poetry run pytest tests/integration/test_ios_device.py -v + +Set only the protocol URL vars for the servers you have available; each protocol +test skips automatically if its URL is not set. `conftest.py` maps this module +to md5 and copies `FILE_CHECKSUM_MD5` into `FILE_CHECKSUM` automatically. + +Include the port in a URL whenever the service does not listen on the default. +IOS honors it, and the driver carries it through to the copy command. + +Environment variables: + IOS_HOST - IP address or hostname of the lab IOS device + IOS_USER - SSH username + IOS_PASS - SSH password + IOS_VRF - Optional VRF name; when set, every copy test routes through this VRF + (needed when the file servers are only reachable via the management VRF). + IOS rejects the vrf keyword on http and https, and the driver omits it + for those two schemes. + FTP_URL - FTP URL of the file to transfer + TFTP_URL - TFTP URL of the file to transfer + SCP_URL - SCP URL of the file to transfer + HTTP_URL - HTTP URL of the file to transfer + HTTPS_URL - HTTPS URL of the file to transfer + SFTP_URL - SFTP URL of the file to transfer + FILE_NAME - Destination filename on the device (default: basename of URL path) + FILE_CHECKSUM_MD5 - Expected md5 checksum of the file (shared across all protocols) + FILE_SIZE - Expected size of the file expressed in FILE_SIZE_UNIT units; used for + the pre-transfer free-space check + FILE_SIZE_UNIT - One of "bytes", "megabytes", or "gigabytes" (default: "bytes") +""" + +import os + +import pytest + +from pyntc.devices import IOSDevice + +from ._helpers import build_file_copy_model + + +@pytest.fixture(scope="module") +def device(): + """Connect to the lab IOS device. Skips all tests if credentials are not set.""" + host = os.environ.get("IOS_HOST") + user = os.environ.get("IOS_USER") + password = os.environ.get("IOS_PASS") + + if not all([host, user, password]): + pytest.skip("IOS_HOST / IOS_USER / IOS_PASS environment variables not set") + + dev = IOSDevice(host, user, password) + yield dev + dev.close() + + +def _build_ios_file_copy_model(env_var): + """Wrap `build_file_copy_model` to stamp `IOS_VRF` onto the model. + + IOS file servers are often only reachable via the management VRF, so the + device's `copy` command needs `vrf ` appended. The shared helper + has no concept of VRF; this driver-local wrapper bridges that gap without + leaking IOS specifics into the shared helper. + """ + model = build_file_copy_model(env_var) + vrf = os.environ.get("IOS_VRF") + if vrf: + model.vrf = vrf + return model + + +def test_device_connects(device): + """Verify the device is reachable and responds to show commands.""" + assert device.hostname + assert device.os_version + + +def test_check_file_exists_false(device, any_file_copy_model): + """Before the copy, the file should not exist (or this test is a no-op if it does).""" + result = device.check_file_exists(any_file_copy_model.file_name) + assert isinstance(result, bool) + + +def test_get_remote_checksum_after_exists(device, any_file_copy_model): + """If the file already exists, verify get_remote_checksum returns a non-empty string.""" + if not device.check_file_exists(any_file_copy_model.file_name): + pytest.skip("File does not exist on device; run test_remote_file_copy_* first") + checksum = device.get_remote_checksum( + any_file_copy_model.file_name, hashing_algorithm=any_file_copy_model.hashing_algorithm + ) + assert checksum and len(checksum) > 0 + + +def test_remote_file_copy_ftp(device): + """Transfer the file using FTP and verify it exists on the device. + + IOS never prompts for FTP credentials. This test fails against a driver that + sends the source URL without them, because the device attempts an anonymous + login and the server refuses it. + """ + model = _build_ios_file_copy_model("FTP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_tftp(device): + """Transfer the file using TFTP and verify it exists on the device.""" + model = _build_ios_file_copy_model("TFTP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_scp(device): + """Transfer the file using SCP and verify it exists on the device. + + IOS prompts for the source username and the password here, so the driver + sends a URL with no credentials and answers the prompts from the model. + """ + model = _build_ios_file_copy_model("SCP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_http(device): + """Transfer the file using HTTP and verify it exists on the device.""" + model = _build_ios_file_copy_model("HTTP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_https(device): + """Transfer the file using HTTPS and verify it exists on the device. + + An IOS release with a dated TLS stack can fail the handshake against a modern + server. That is a device and server mismatch rather than a driver defect. + """ + model = _build_ios_file_copy_model("HTTPS_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) + + +def test_remote_file_copy_sftp(device): + """Transfer the file using SFTP and verify it exists on the device.""" + model = _build_ios_file_copy_model("SFTP_URL") + device.remote_file_copy(model) + assert device.check_file_exists(model.file_name) diff --git a/tests/unit/test_devices/test_ios_device.py b/tests/unit/test_devices/test_ios_device.py index a8f68e97..6e50d32a 100644 --- a/tests/unit/test_devices/test_ios_device.py +++ b/tests/unit/test_devices/test_ios_device.py @@ -694,6 +694,138 @@ def test_remote_file_copy_skips_space_check_when_file_size_omitted(self, mock_ch mock_check_free_space.assert_not_called() self.device.native.send_command.assert_called() + @mock.patch.object(IOSDevice, "verify_file") + def test_remote_file_copy_ftp_embeds_credentials(self, mock_verify): + """IOS reads FTP credentials from the URL and never prompts for them.""" + src = FileCopyModel( + download_url="ftp://10.1.100.220/IOS-XE/test.bin", + checksum="12345", + file_name="test.bin", + hashing_algorithm="md5", + timeout=900, + username="ntc", + token="ntc1234", + ) + mock_verify.side_effect = [False, True] + self.device.native.send_command.return_value = "94038 bytes copied in 0.357 secs" + self.device.native.find_prompt.return_value = "Router#" + + self.device.remote_file_copy(src, file_system="flash:") + + self.device.native.send_command.assert_called_once_with( + "copy ftp://ntc:ntc1234@10.1.100.220/IOS-XE/test.bin flash:test.bin", + expect_string=mock.ANY, + read_timeout=900, + ) + + @mock.patch.object(IOSDevice, "verify_file") + def test_remote_file_copy_ftp_keeps_non_default_port(self, mock_verify): + """A non-default port on the source URL survives into the copy command.""" + src = FileCopyModel( + download_url="ftp://10.1.100.220:2121/IOS-XE/test.bin", + checksum="12345", + file_name="test.bin", + hashing_algorithm="md5", + timeout=900, + username="ntc", + token="ntc1234", + ) + mock_verify.side_effect = [False, True] + self.device.native.send_command.return_value = "94038 bytes copied in 0.357 secs" + self.device.native.find_prompt.return_value = "Router#" + + self.device.remote_file_copy(src, file_system="flash:") + + self.device.native.send_command.assert_called_once_with( + "copy ftp://ntc:ntc1234@10.1.100.220:2121/IOS-XE/test.bin flash:test.bin", + expect_string=mock.ANY, + read_timeout=900, + ) + + @mock.patch.object(IOSDevice, "verify_file") + def test_remote_file_copy_scp_keeps_bare_url_and_walks_prompts(self, mock_verify): + """IOS prompts for SCP credentials, so its URL stays free of them.""" + src = FileCopyModel( + download_url="scp://10.1.100.220:2022/IOS-XE/test.bin", + checksum="12345", + file_name="test.bin", + hashing_algorithm="md5", + timeout=900, + username="ntc", + token="ntc1234", + ) + mock_verify.side_effect = [False, True] + self.device.native.send_command.side_effect = [ + "Source username [ntc]?", + "Password:", + "94038 bytes copied in 3.017 secs", + ] + self.device.native.find_prompt.return_value = "Router#" + + self.device.remote_file_copy(src, file_system="flash:") + + self.device.native.send_command.assert_has_calls( + [ + mock.call( + "copy scp://10.1.100.220:2022/IOS-XE/test.bin flash:test.bin", + expect_string=mock.ANY, + read_timeout=900, + ), + mock.call("ntc", expect_string=mock.ANY, read_timeout=900, cmd_verify=True), + mock.call("ntc1234", expect_string=mock.ANY, read_timeout=900, cmd_verify=False), + ] + ) + + @mock.patch.object(IOSDevice, "verify_file") + def test_remote_file_copy_appends_vrf_for_ftp(self, mock_verify): + """The copy command carries the VRF for schemes whose parser accepts it.""" + src = FileCopyModel( + download_url="ftp://10.1.100.220/IOS-XE/test.bin", + checksum="12345", + file_name="test.bin", + hashing_algorithm="md5", + timeout=900, + username="ntc", + token="ntc1234", + vrf="Mgmt-vrf", + ) + mock_verify.side_effect = [False, True] + self.device.native.send_command.return_value = "94038 bytes copied in 0.357 secs" + self.device.native.find_prompt.return_value = "Router#" + + self.device.remote_file_copy(src, file_system="flash:") + + self.device.native.send_command.assert_called_once_with( + "copy ftp://ntc:ntc1234@10.1.100.220/IOS-XE/test.bin flash:test.bin vrf Mgmt-vrf", + expect_string=mock.ANY, + read_timeout=900, + ) + + @mock.patch.object(IOSDevice, "verify_file") + def test_remote_file_copy_omits_vrf_for_http(self, mock_verify): + """The HTTP copy command rejects a trailing vrf keyword, so it is never added.""" + src = FileCopyModel( + download_url="http://10.1.100.220:8081/IOS-XE/test.bin", + checksum="12345", + file_name="test.bin", + hashing_algorithm="md5", + timeout=900, + username="ntc", + token="ntc1234", + vrf="Mgmt-vrf", + ) + mock_verify.side_effect = [False, True] + self.device.native.send_command.return_value = "94038 bytes copied in 0.357 secs" + self.device.native.find_prompt.return_value = "Router#" + + self.device.remote_file_copy(src, file_system="flash:") + + self.device.native.send_command.assert_called_once_with( + "copy http://ntc:ntc1234@10.1.100.220:8081/IOS-XE/test.bin flash:test.bin", + expect_string=mock.ANY, + read_timeout=900, + ) + if __name__ == "__main__": unittest.main() From 04523e0d904560c79243dc02da40ec074b511902 Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Tue, 15 Sep 2026 15:35:28 -0400 Subject: [PATCH 2/2] Report the device error when a remote file copy fails --- changes/429.fixed.1 | 1 + changes/429.fixed.2 | 1 + pyntc/devices/asa_device.py | 22 ++++++--- pyntc/devices/ios_device.py | 30 ++++++++++--- pyntc/devices/iosxr_device.py | 17 ++++--- pyntc/devices/nxos_device.py | 25 +++++++++-- tests/unit/test_devices/test_asa_device.py | 27 +++++++++++ tests/unit/test_devices/test_ios_device.py | 47 ++++++++++++++++++++ tests/unit/test_devices/test_iosxr_device.py | 16 +++++++ tests/unit/test_devices/test_nxos_device.py | 41 +++++++++++++++++ 10 files changed, 209 insertions(+), 18 deletions(-) create mode 100644 changes/429.fixed.1 create mode 100644 changes/429.fixed.2 diff --git a/changes/429.fixed.1 b/changes/429.fixed.1 new file mode 100644 index 00000000..ab0274c0 --- /dev/null +++ b/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. diff --git a/changes/429.fixed.2 b/changes/429.fixed.2 new file mode 100644 index 00000000..03b80f4d --- /dev/null +++ b/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. diff --git a/pyntc/devices/asa_device.py b/pyntc/devices/asa_device.py index b147b9b4..d16b21d3 100644 --- a/pyntc/devices/asa_device.py +++ b/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): diff --git a/pyntc/devices/ios_device.py b/pyntc/devices/ios_device.py index a257af5c..3b5dc10c 100644 --- a/pyntc/devices/ios_device.py +++ b/pyntc/devices/ios_device.py @@ -810,6 +810,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: FileCopyModel, file_system: str, dest: str) -> str: """Build the copy command for transfers where IOS prompts for the credentials it needs. @@ -888,8 +893,9 @@ def remote_file_copy( # noqa: R0912 pylint: disable=too-many-branches 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 @@ -897,16 +903,30 @@ def remote_file_copy( # noqa: R0912 pylint: disable=too-many-branches 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): diff --git a/pyntc/devices/iosxr_device.py b/pyntc/devices/iosxr_device.py index 2dc236de..de941b21 100644 --- a/pyntc/devices/iosxr_device.py +++ b/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) diff --git a/pyntc/devices/nxos_device.py b/pyntc/devices/nxos_device.py index 8583cbbc..06a4f8a8 100644 --- a/pyntc/devices/nxos_device.py +++ b/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,8 +621,9 @@ 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 @@ -623,6 +631,17 @@ def remote_file_copy(self, src: FileCopyModel, dest=None, file_system=None, **kw 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( diff --git a/tests/unit/test_devices/test_asa_device.py b/tests/unit/test_devices/test_asa_device.py index ae647423..1eb88163 100644 --- a/tests/unit/test_devices/test_asa_device.py +++ b/tests/unit/test_devices/test_asa_device.py @@ -1154,6 +1154,33 @@ def test_remote_file_copy_error_in_output(mock_verify, mock_fs, asa_device): asa_device.remote_file_copy(FILE_COPY_MODEL_FTP) +@mock.patch.object(ASADevice, "_get_file_system", return_value="disk0:") +@mock.patch.object(ASADevice, "verify_file", return_value=False) +def test_remote_file_copy_error_carries_device_output(mock_verify, mock_fs, asa_device): + """The raised error repeats what the device said, with the token removed.""" + asa_device.native.find_prompt.return_value = "asa5512#" + asa_device.native.send_command.return_value = ( + "%Error opening ftp://example-user:example-password@192.0.2.1/asa.bin (Incorrect Login/Password)" + ) + with pytest.raises(FileTransferError) as err: + asa_device.remote_file_copy(FILE_COPY_MODEL_FTP) + + assert "Incorrect Login/Password" in err.value.message + assert "example-password" not in err.value.message + + +@mock.patch.object(ASADevice, "_get_file_system", return_value="disk0:") +@mock.patch.object(ASADevice, "verify_file", return_value=False) +def test_remote_file_copy_unrecognized_output_carries_device_output(mock_verify, mock_fs, asa_device): + """The guard against an endless prompt loop names what the device actually sent.""" + asa_device.native.find_prompt.return_value = "asa5512#" + asa_device.native.send_command.return_value = "something the driver has never seen" + with pytest.raises(FileTransferError) as err: + asa_device.remote_file_copy(FILE_COPY_MODEL_FTP) + + assert "Unexpected output" in err.value.message + + @mock.patch.object(ASADevice, "_get_file_system", return_value="disk0:") @mock.patch.object(ASADevice, "verify_file", side_effect=[False, False]) def test_remote_file_copy_verify_fails_after_copy(mock_verify, mock_fs, asa_device): diff --git a/tests/unit/test_devices/test_ios_device.py b/tests/unit/test_devices/test_ios_device.py index 6e50d32a..d075702a 100644 --- a/tests/unit/test_devices/test_ios_device.py +++ b/tests/unit/test_devices/test_ios_device.py @@ -826,6 +826,53 @@ def test_remote_file_copy_omits_vrf_for_http(self, mock_verify): read_timeout=900, ) + @mock.patch.object(IOSDevice, "verify_file") + def test_remote_file_copy_error_carries_device_output(self, mock_verify): + """The raised error repeats what the device said, with the token removed.""" + from pyntc.errors import FileTransferError + + src = FileCopyModel( + download_url="ftp://10.1.100.220/IOS-XE/test.bin", + checksum="12345", + file_name="test.bin", + hashing_algorithm="md5", + username="ntc", + token="ntc1234", + ) + mock_verify.return_value = False + self.device.native.find_prompt.return_value = "Router#" + self.device.native.send_command.return_value = ( + "%Error opening ftp://ntc:ntc1234@10.1.100.220/IOS-XE/test.bin (Incorrect Login/Password)" + ) + + with self.assertRaises(FileTransferError) as err: + self.device.remote_file_copy(src, file_system="flash:") + + self.assertIn("Incorrect Login/Password", err.exception.message) + self.assertNotIn("ntc1234", err.exception.message) + + @mock.patch.object(IOSDevice, "verify_file") + def test_remote_file_copy_raises_on_unrecognized_output(self, mock_verify): + """Output matching no prompt and no marker raises instead of looping forever.""" + from pyntc.errors import FileTransferError + + src = FileCopyModel( + download_url="ftp://10.1.100.220/IOS-XE/test.bin", + checksum="12345", + file_name="test.bin", + hashing_algorithm="md5", + username="ntc", + token="ntc1234", + ) + mock_verify.return_value = False + self.device.native.find_prompt.return_value = "Router#" + self.device.native.send_command.return_value = "something the driver has never seen" + + with self.assertRaises(FileTransferError) as err: + self.device.remote_file_copy(src, file_system="flash:") + + self.assertIn("Unexpected output", err.exception.message) + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_devices/test_iosxr_device.py b/tests/unit/test_devices/test_iosxr_device.py index 8eeb41c9..74ef770c 100644 --- a/tests/unit/test_devices/test_iosxr_device.py +++ b/tests/unit/test_devices/test_iosxr_device.py @@ -573,6 +573,22 @@ def test_remote_file_copy_error_raises(self, *_mocks): with self.assertRaises(FileTransferError): self.device.remote_file_copy(src) + @mock.patch.object(IOSXRDevice, "check_file_exists", side_effect=[False]) + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") + def test_remote_file_copy_error_carries_device_output(self, *_mocks): + """The raised error repeats what the device said, with the token removed.""" + self.device.native.find_prompt.return_value = PROMPT + self.device.native.send_command.return_value = ( + "%Error opening ftp://ntc:ntc1234@192.0.2.1/image.iso (Incorrect Login/Password)" + ) + src = FileCopyModel(download_url=ISO_URL, checksum="", file_name=ISO, username="ntc", token="ntc1234") + + with self.assertRaises(FileTransferError) as err: + self.device.remote_file_copy(src) + + self.assertIn("Incorrect Login/Password", err.exception.message) + self.assertNotIn("ntc1234", err.exception.message) + @mock.patch.object(IOSXRDevice, "_get_file_system", return_value="harddisk:") def test_get_remote_checksum_md5(self, *_mocks): self.device.native.send_command.return_value = RUN_MD5SUM diff --git a/tests/unit/test_devices/test_nxos_device.py b/tests/unit/test_devices/test_nxos_device.py index a8e9e8b4..fb6c4625 100644 --- a/tests/unit/test_devices/test_nxos_device.py +++ b/tests/unit/test_devices/test_nxos_device.py @@ -514,6 +514,47 @@ def test_remote_file_copy_transfer_success(self): call_args = self.device.native_ssh.send_command.call_args self.assertIn("expect_string", call_args.kwargs) + def test_remote_file_copy_error_carries_device_output(self): + """The raised error repeats what the device said, with the token removed.""" + src = FileCopyModel( + download_url="ftp://example.com/nxos.bin", + checksum="abc123", + file_name="nxos.bin", + hashing_algorithm="md5", + timeout=30, + username="ntc", + token="ntc1234", + ) + self.device.native_ssh.find_prompt.return_value = "host#" + self.device.native_ssh.send_command.side_effect = None + self.device.native_ssh.send_command.return_value = ( + "%Error opening ftp://ntc:ntc1234@example.com/nxos.bin (Incorrect Login/Password)" + ) + with mock.patch.object(NXOSDevice, "verify_file", return_value=False): + with self.assertRaises(FileTransferError) as err: + self.device.remote_file_copy(src, file_system="bootflash:") + + self.assertIn("Incorrect Login/Password", err.exception.message) + self.assertNotIn("ntc1234", err.exception.message) + + def test_remote_file_copy_raises_on_unrecognized_output(self): + """Output matching no prompt and no marker raises instead of looping forever.""" + src = FileCopyModel( + download_url="ftp://example.com/nxos.bin", + checksum="abc123", + file_name="nxos.bin", + hashing_algorithm="md5", + timeout=30, + ) + self.device.native_ssh.find_prompt.return_value = "host#" + self.device.native_ssh.send_command.side_effect = None + self.device.native_ssh.send_command.return_value = "something the driver has never seen" + with mock.patch.object(NXOSDevice, "verify_file", return_value=False): + with self.assertRaises(FileTransferError) as err: + self.device.remote_file_copy(src, file_system="bootflash:") + + self.assertIn("Unexpected output", err.exception.message) + def test_remote_file_copy_transfer_fails_verification(self): src = FileCopyModel( download_url="http://example.com/nxos.bin",