From 0d45c521d2ff12cf49d00bf1fc485648a66792f4 Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Tue, 15 Sep 2026 15:33:38 -0400 Subject: [PATCH 1/4] 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/4] 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", From 6c2475daf58e0695025a8499c0b2e8a4f9b0bf36 Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Tue, 22 Sep 2026 09:54:27 -0400 Subject: [PATCH 3/4] Use netmiko SecretsFilter to scrub IOS copy credentials --- changes/429.security | 1 + pyntc/devices/ios_device.py | 132 +++++++++++++++------ tests/unit/test_devices/test_ios_device.py | 89 ++++++++++++++ 3 files changed, 187 insertions(+), 35 deletions(-) create mode 100644 changes/429.security diff --git a/changes/429.security b/changes/429.security new file mode 100644 index 00000000..6e630616 --- /dev/null +++ b/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. diff --git a/pyntc/devices/ios_device.py b/pyntc/devices/ios_device.py index 3b5dc10c..72d438dd 100644 --- a/pyntc/devices/ios_device.py +++ b/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 @@ -41,6 +42,60 @@ 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 (BaseConnection): 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 @@ -812,7 +867,10 @@ def _source_path(src: FileCopyModel, dest: str) -> str: @staticmethod def _mask_token(output: str, src: FileCopyModel) -> str: - """Replace the token in device output, so it is safe to log or raise.""" + """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: @@ -834,7 +892,7 @@ def _build_url_copy_command_with_creds(self, src: FileCopyModel, file_system: st 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 + 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. @@ -873,47 +931,51 @@ def remote_file_copy( # noqa: R0912 pylint: disable=too-many-branches keys = list(prompt_answers.keys()) + [re.escape(current_prompt)] expect_regex = f"({'|'.join(keys)})" - if src.username and src.scheme in IOS_URL_CREDENTIAL_SCHEMES: + 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): - 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 + 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 - 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 diff --git a/tests/unit/test_devices/test_ios_device.py b/tests/unit/test_devices/test_ios_device.py index d075702a..aab2a97d 100644 --- a/tests/unit/test_devices/test_ios_device.py +++ b/tests/unit/test_devices/test_ios_device.py @@ -1,9 +1,11 @@ +import logging import os import time import unittest import mock import pytest +from netmiko.base_connection import SecretsFilter from pyntc.devices import IOSDevice from pyntc.devices import ios_device as ios_module @@ -74,6 +76,8 @@ def setUp(self, mock_miko, mock_close, mock_open): mock_miko.send_command_timing.side_effect = send_command mock_miko.send_command_expect.side_effect = send_command_expect + # Match a real connection, whose filter is seeded with the device password. + mock_miko._secrets_filter = SecretsFilter(no_log={"password": "pass"}) self.device.native = mock_miko def tearDown(self): @@ -851,6 +855,91 @@ def test_remote_file_copy_error_carries_device_output(self, mock_verify): 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_hides_url_token_from_logs(self, mock_verify): + """The token in the copy URL is scrubbed from log records and unregistered afterwards.""" + 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)" + ) + pyntc_log = ios_module.log.get_log() + + with self.assertLogs(pyntc_log, level=logging.ERROR) as captured: + with self.assertRaises(FileTransferError): + self.device.remote_file_copy(src, file_system="flash:") + + logged = "\n".join(captured.output) + self.assertIn("Incorrect Login/Password", logged) + self.assertNotIn("ntc1234", logged) + self.assertEqual(pyntc_log.filters, []) + self.assertNotIn("file_copy_token", self.device.native._secrets_filter.no_log) + + @mock.patch.object(IOSDevice, "verify_file") + def test_remote_file_copy_ftp_registers_token_with_netmiko(self, mock_verify): + """The token joins netmiko's no_log mapping while the copy runs, and leaves afterwards.""" + 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.side_effect = [False, True] + no_log = self.device.native._secrets_filter.no_log + registered = {} + + def capture_no_log(*args, **kwargs): + registered.update(no_log) + return "94038 bytes copied in 0.357 secs" + + self.device.native.find_prompt.return_value = "Router#" + self.device.native.send_command.side_effect = capture_no_log + + self.device.remote_file_copy(src, file_system="flash:") + + self.assertEqual(registered, {"password": "pass", "file_copy_token": "ntc1234"}) + self.assertEqual(no_log, {"password": "pass"}) + + @mock.patch.object(IOSDevice, "verify_file") + def test_remote_file_copy_scp_keeps_token_out_of_netmiko(self, mock_verify): + """SCP answers a prompt for its password, so the token never joins the mapping.""" + src = FileCopyModel( + download_url="scp://10.1.100.220:2022/IOS-XE/test.bin", + checksum="12345", + file_name="test.bin", + hashing_algorithm="md5", + username="ntc", + token="ntc1234", + ) + mock_verify.side_effect = [False, True] + no_log = self.device.native._secrets_filter.no_log + registered = {} + responses = iter(["Source username [ntc]?", "Password:", "94038 bytes copied in 3.017 secs"]) + + def capture_no_log(*args, **kwargs): + registered.update(no_log) + return next(responses) + + self.device.native.find_prompt.return_value = "Router#" + self.device.native.send_command.side_effect = capture_no_log + + self.device.remote_file_copy(src, file_system="flash:") + + self.assertEqual(registered, {"password": "pass"}) + self.assertEqual(no_log, {"password": "pass"}) + @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 0ad67950bf2b11b523484692b1ae0cd8e11f59e6 Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Thu, 24 Sep 2026 09:11:41 -0400 Subject: [PATCH 4/4] review feedback --- changes/429.changed | 1 + poetry.lock | 13 ++- pyntc/devices/ios_device.py | 69 +++++++++---- pyproject.toml | 2 +- tests/unit/test_devices/test_ios_device.py | 111 +++++++++++++++++++++ 5 files changed, 172 insertions(+), 24 deletions(-) create mode 100644 changes/429.changed diff --git a/changes/429.changed b/changes/429.changed new file mode 100644 index 00000000..a18c3e0c --- /dev/null +++ b/changes/429.changed @@ -0,0 +1 @@ +Raised the minimum supported netmiko version to 4.4. diff --git a/poetry.lock b/poetry.lock index 5b622e1c..222a8460 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.3.1 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.4.0 and should not be changed by hand. [[package]] name = "astroid" @@ -876,6 +876,7 @@ files = [ {file = "lxml-6.1.1-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:37a58976370f36d9329d118ad0b953c5aeb9119ac9c6a4e258942a225d0573a1"}, {file = "lxml-6.1.1-cp310-cp310-win32.whl", hash = "sha256:cea3f4c1af79af13cdb2da0c028111d8f8522d4f22a000c82385535f24e5cf3a"}, {file = "lxml-6.1.1-cp310-cp310-win_amd64.whl", hash = "sha256:3abf332af33a74288675d936fe861fd4344da0dd6622193fbc4f2bfbb35536b5"}, + {file = "lxml-6.1.1-cp310-cp310-win_arm64.whl", hash = "sha256:8dadbe5b217ff35b6a8d16610dd710219b59b76d13f0e3f0d9f36786206e4485"}, {file = "lxml-6.1.1-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:53b7d2b7a10b1c35c0a5e21e9224accf60c1bbfba523990732e521b2b73adef2"}, {file = "lxml-6.1.1-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:ff3f333630ab480244a1bff72043e511a91eb22e7595dead8653ee5612dd8f3d"}, {file = "lxml-6.1.1-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:a4bbea04c97f6d78a48e3fbc1cb9116d2780b1b39e03a23f6eb9b603fd61f510"}, @@ -891,6 +892,7 @@ files = [ {file = "lxml-6.1.1-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:80c2dfadb855da477cf73373ad29a333535dedb9b12bad02c9814c8e2b43bf08"}, {file = "lxml-6.1.1-cp311-cp311-win32.whl", hash = "sha256:30a89d3ac8faec007453fb541f3f46807eeec88edd5826f6e3fe001752a2c621"}, {file = "lxml-6.1.1-cp311-cp311-win_amd64.whl", hash = "sha256:abbefa31eee84842140f67acef1c828e28bba8bbf0c3bc6e5492a9af88152c28"}, + {file = "lxml-6.1.1-cp311-cp311-win_arm64.whl", hash = "sha256:dcb292aa7fe485ceff7af4f92e46c5af397daec5dff64871a528f0fc47a3cc5b"}, {file = "lxml-6.1.1-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:104c09bda8d2a562824c0e319d0768ce26a779b7601e0931d33b09b53c392ef7"}, {file = "lxml-6.1.1-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:25c6997a9a534e016695a0ba06b2f07945de682731ff01065b6d5a4474179da1"}, {file = "lxml-6.1.1-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:c921ba5c51e4e9f63b8b00267d06566e1f63407408a0496da2d1d0bfc819c7fc"}, @@ -908,6 +910,7 @@ files = [ {file = "lxml-6.1.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:aa366a1e55b8ebfe8ca8ddc3cfe75c8ebade181aeb0f661d0cb05986b647f72a"}, {file = "lxml-6.1.1-cp312-cp312-win32.whl", hash = "sha256:126c93f7f56f0eda92f6d8c619edc463a4f23d9252f1c9d0405a76f25fa9f11a"}, {file = "lxml-6.1.1-cp312-cp312-win_amd64.whl", hash = "sha256:26e6eda8d38c1fcab1090dd196ee87cbd13788e531937610e2589085de074e77"}, + {file = "lxml-6.1.1-cp312-cp312-win_arm64.whl", hash = "sha256:6540377fbd53fe1b629172288c464fb18db11ce1fa7dc15891da10aa9dcc3e7f"}, {file = "lxml-6.1.1-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:68a9198d0fc122d14bb76837de9aa80cf84caed990b5b237f532ed87d3706736"}, {file = "lxml-6.1.1-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:7d47866cb32fb503450b6edc9df355d10dc49836af2e89901bd6ac6b0896d9d9"}, {file = "lxml-6.1.1-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:eb7c9811bfaa8b1ed5ed319f5d370dfbcaa59d52ea64be2a5a85e18195930354"}, @@ -925,6 +928,7 @@ files = [ {file = "lxml-6.1.1-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:47402e62c52ff5988c1e8c6c63177f5708bccf48e366dea4e3dcf1e645e04947"}, {file = "lxml-6.1.1-cp313-cp313-win32.whl", hash = "sha256:3483644525531e1d5762b0c44a8e18b6efba321b6dcf8a8952de10b037618bca"}, {file = "lxml-6.1.1-cp313-cp313-win_amd64.whl", hash = "sha256:a10bd2fd62e8ce916ececb342f348f190724a098c1faa056fdfb2a22ad5e8660"}, + {file = "lxml-6.1.1-cp313-cp313-win_arm64.whl", hash = "sha256:424aa57aca0897eb922aef34395bd1289b3b6f04e6bae20ea123c0c7e333cffc"}, {file = "lxml-6.1.1-cp314-cp314-macosx_10_15_universal2.whl", hash = "sha256:19b7ab10b210b0b3ad7985d9ac4eb66ab09a90b20fe6e2f7ba55d01a234345d0"}, {file = "lxml-6.1.1-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:c08e5c694306507275f2290073350c4f32e383db15213b2c69e7ff39c1193840"}, {file = "lxml-6.1.1-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:74a9717fd0d82effef5c2854f0d917231d5324b5a3eb7275c43ac9fa32f97a14"}, @@ -942,6 +946,7 @@ files = [ {file = "lxml-6.1.1-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:63876be28efefa04a1df615b46770e82042cce445cfdce55160522f57b231ccb"}, {file = "lxml-6.1.1-cp314-cp314-win32.whl", hash = "sha256:7f7a92e8583f06b1fd49d01158143b8461cfcd135dcb10ec807270a3051bd603"}, {file = "lxml-6.1.1-cp314-cp314-win_amd64.whl", hash = "sha256:b2d444f2e66624d68e9c6b211e28a76e22fff5fcabcfff4deac18b529b7d4137"}, + {file = "lxml-6.1.1-cp314-cp314-win_arm64.whl", hash = "sha256:3fd9728a2735fda14f4e8235830c86b539e9661e849665bf926d3f867943b4bf"}, {file = "lxml-6.1.1-cp314-cp314t-macosx_10_15_universal2.whl", hash = "sha256:787b2496d0dbe8cd180984e8d29e3a6f76e7ea34db781cb3bd55e4ba1ef8b4ee"}, {file = "lxml-6.1.1-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:2c8daa471358dc2d6fcf02165e80ec68f77871a286df95bc5cc3816153b0fd2c"}, {file = "lxml-6.1.1-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:acd7d70b64c0aae0c7922cca83d288a16f5f6da523637697872253415269baef"}, @@ -959,6 +964,7 @@ files = [ {file = "lxml-6.1.1-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:9e36f163528fc50cbef305f02a5fd66d404edf7049cdaff211dbc2cba5a7013e"}, {file = "lxml-6.1.1-cp314-cp314t-win32.whl", hash = "sha256:649dda677cf3bd6ac9ae14007ba0c824ded8ce5808b53fc7431d9140399118c1"}, {file = "lxml-6.1.1-cp314-cp314t-win_amd64.whl", hash = "sha256:793033d6c5cdf33a573f910d9bea14ef8f5771820411d118da8e1182edb53d5e"}, + {file = "lxml-6.1.1-cp314-cp314t-win_arm64.whl", hash = "sha256:58bb955caba94e467d2a96da17660d2d704e0675894cba21ab8a775b8621fd1c"}, {file = "lxml-6.1.1-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:6689e828a94eee4f139408c337bb198e014724bb8a8c26d3cfac49d119ed69a6"}, {file = "lxml-6.1.1-cp38-cp38-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:bdebcc8a75d38c7598dfb2c9ed852d7a9eb4a10d6e2d0764b919b802bf32ac88"}, {file = "lxml-6.1.1-cp38-cp38-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8be8ad51249698103d24b0571df35a10990fbe93dd043b6c024172189485f5e3"}, @@ -981,6 +987,7 @@ files = [ {file = "lxml-6.1.1-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:c9a4b821dc7055bf9e05ff5719e18ec501f75c0f0bbfabd573b277559780833d"}, {file = "lxml-6.1.1-cp39-cp39-win32.whl", hash = "sha256:639f6c857d91d9be29bd7502348d6736dab168b54b5158cd899abf11684dc186"}, {file = "lxml-6.1.1-cp39-cp39-win_amd64.whl", hash = "sha256:34c2d737beabfe35baada43941ed519251e9a12e779031496bcd5d539fcfd730"}, + {file = "lxml-6.1.1-cp39-cp39-win_arm64.whl", hash = "sha256:07a4a68e286ee7a1ed7dfb8af83e615757c0ccfe9f18c6b4ea6771388d9ba8c9"}, {file = "lxml-6.1.1-pp311-pypy311_pp73-macosx_10_15_x86_64.whl", hash = "sha256:31033dc34636ea6b7d5cc11b1ddbda78a14de858ba9d3e1ed4b69a3085bc521e"}, {file = "lxml-6.1.1-pp311-pypy311_pp73-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:3893c14c4b6ac5b2d54ba8cf03e99fe5104e592de491f19bd6b82756c09f8004"}, {file = "lxml-6.1.1-pp311-pypy311_pp73-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:c07da4cebf6889f03ebac8d238f62318e29f495de0aa18a51ea14e61ae907e2e"}, @@ -1681,8 +1688,8 @@ astroid = ">=3.3.8,<=3.4.0.dev0" colorama = {version = ">=0.4.5", markers = "sys_platform == \"win32\""} dill = [ {version = ">=0.2", markers = "python_version < \"3.11\""}, - {version = ">=0.3.7", markers = "python_version >= \"3.12\""}, {version = ">=0.3.6", markers = "python_version == \"3.11\""}, + {version = ">=0.3.7", markers = "python_version >= \"3.12\""}, ] isort = ">=4.2.5,<5.13 || >5.13,<7" mccabe = ">=0.6,<0.8" @@ -2363,4 +2370,4 @@ test = ["coverage", "hypothesis", "pytest"] [metadata] lock-version = "2.1" python-versions = ">=3.10,<3.15" -content-hash = "7a9d5e519fffc843c55f428243fa9f3083d6027e1dd5b94d2714831da8e0a1fc" +content-hash = "cce168ee475f01976d34653679a7d99fa2cb26fbb9674e1b11641da85964ac33" diff --git a/pyntc/devices/ios_device.py b/pyntc/devices/ios_device.py index 72d438dd..057d4f2b 100644 --- a/pyntc/devices/ios_device.py +++ b/pyntc/devices/ios_device.py @@ -4,6 +4,8 @@ import re import time import warnings +from typing import Optional +from urllib.parse import ParseResult, quote, urlparse from netmiko import ConnectHandler, FileTransfer from netmiko.base_connection import SecretsFilter @@ -56,44 +58,46 @@ class HideUrlCredentials: 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. + def __init__(self, native, secrets): + """Capture the registries the secrets are added to and removed from. Args: - native (BaseConnection): 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. + native (BaseConnection): The netmiko connection whose no_log mapping the secrets join. + secrets (dict): Values to hide, keyed by the name each takes in netmiko's no_log + mapping. An empty mapping is a no-op. """ - self.token = token + self.secrets = secrets 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. + """Add a `SecretsFilter` to the pyntc logger and the secrets 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 + its own `SecretsFilter` and its `SessionLog` read from, so writing the secrets there covers the copy command as netmiko sends it. Returns: (HideUrlCredentials): This instance. """ - if not self.token: + if not self.secrets: 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_filter = SecretsFilter(no_log=self.secrets) + self.netmiko_no_log.update(self.secrets) self.pyntc_log.addFilter(self.pyntc_filter) return self def __exit__(self, exc_type, exc_value, traceback): - """Unregister the token. + """Unregister the secrets. 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) + for key in self.secrets: + self.netmiko_no_log.pop(key, None) self.pyntc_log.removeFilter(self.pyntc_filter) return False @@ -861,17 +865,40 @@ def _netloc(src: FileCopyModel) -> str: 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 _source_path(parsed: ParseResult, dest: str) -> str: + """Return the path and query from the parsed URL, falling back to dest when the path is empty.""" + path = parsed.path if parsed.path and parsed.path != "/" else f"/{dest}" + return f"{path}?{parsed.query}" if parsed.query else path @staticmethod - def _mask_token(output: str, src: FileCopyModel) -> str: + def _url_credential(value: str, url_value: Optional[str]) -> str: + """Return a credential as it must appear in a copy URL, percent-encoded exactly once. + + A value equal to the one `urlparse` read from the download URL is already encoded + and passes through. Any other value arrived as an argument in plain text. + """ + return value if value == url_value else quote(value, safe="") + + @classmethod + def _url_secrets(cls, src: FileCopyModel) -> dict: + """Return the copy URL password in each form that can reach a log, keyed for netmiko's no_log.""" + if not src.token: + return {} + secrets = {FILE_COPY_NO_LOG_KEY: src.token} + encoded = cls._url_credential(src.token, urlparse(src.download_url).password) + if encoded != src.token: + secrets[f"{FILE_COPY_NO_LOG_KEY}_encoded"] = encoded + return secrets + + @classmethod + def _mask_token(cls, 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 + for secret in cls._url_secrets(src).values(): + output = output.replace(secret, "*****") + return 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. @@ -887,9 +914,11 @@ def _build_url_copy_command_with_creds(self, src: FileCopyModel, file_system: st FTP, HTTP and HTTPS never prompt. A URL without credentials makes the device attempt an anonymous login, which the server rejects. """ + parsed = urlparse(src.download_url) netloc = self._netloc(src) - path = self._source_path(src, dest) - credentials = f"{src.username}:{src.token}" if src.token else src.username + path = self._source_path(parsed, dest) + username = self._url_credential(src.username, parsed.username) + credentials = f"{username}:{self._url_credential(src.token, parsed.password)}" if src.token else 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 @@ -939,7 +968,7 @@ def remote_file_copy( # noqa: R0912 pylint: disable=too-many-branches,too-many- if src.vrf and src.scheme not in IOS_NO_VRF_SCHEMES: command = f"{command} vrf {src.vrf}" - with HideUrlCredentials(self.native, src.token if credentials_in_url else None): + with HideUrlCredentials(self.native, self._url_secrets(src) if credentials_in_url else {}): # _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) diff --git a/pyproject.toml b/pyproject.toml index 2cfa2f59..b17d435d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,7 +27,7 @@ include = [ python = ">=3.10,<3.15" f5-sdk = "^3.0.21" junos-eznc = ">=2.0,<3.0" -netmiko = ">=4.0,<5.0" +netmiko = ">=4.4,<5.0" pyeapi = ">=1.0,<2.0" requests = ">=2.0,<3.0" scp = ">=0.15,<1.0" diff --git a/tests/unit/test_devices/test_ios_device.py b/tests/unit/test_devices/test_ios_device.py index aab2a97d..1b45ba0e 100644 --- a/tests/unit/test_devices/test_ios_device.py +++ b/tests/unit/test_devices/test_ios_device.py @@ -746,6 +746,117 @@ def test_remote_file_copy_ftp_keeps_non_default_port(self, mock_verify): read_timeout=900, ) + @mock.patch.object(IOSDevice, "verify_file") + def test_remote_file_copy_ftp_keeps_query_string(self, mock_verify): + """A query string on the source URL survives into the copy command.""" + src = FileCopyModel( + download_url="ftp://10.1.100.220/IOS-XE/test.bin?ver=2", + 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?ver=2 flash:test.bin", + expect_string=mock.ANY, + read_timeout=900, + ) + + @mock.patch.object(IOSDevice, "verify_file") + def test_remote_file_copy_ftp_encodes_argument_credentials(self, mock_verify): + """Credentials passed as arguments are percent-encoded before they join the URL.""" + 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="us@er", + token="p@ss/w0rd", + ) + 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://us%40er:p%40ss%2Fw0rd@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_url_credentials_encoded_once(self, mock_verify): + """Credentials read from the URL are already encoded and are not encoded again.""" + src = FileCopyModel( + download_url="ftp://us%40er:p%40ss@10.1.100.220/IOS-XE/test.bin", + checksum="12345", + file_name="test.bin", + hashing_algorithm="md5", + timeout=900, + ) + 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://us%40er:p%40ss@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_hides_encoded_token(self, mock_verify): + """The token is hidden in both its plain and its encoded form.""" + 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="p@ss/w0rd", + ) + mock_verify.return_value = False + no_log = self.device.native._secrets_filter.no_log + registered = {} + + def capture_no_log(*args, **kwargs): + registered.update(no_log) + return "%Error opening ftp://ntc:p%40ss%2Fw0rd@10.1.100.220/IOS-XE/test.bin (Incorrect Login/Password)" + + self.device.native.find_prompt.return_value = "Router#" + self.device.native.send_command.side_effect = capture_no_log + pyntc_log = ios_module.log.get_log() + + with self.assertLogs(pyntc_log, level=logging.ERROR) as captured: + with self.assertRaises(FileTransferError) as raised: + self.device.remote_file_copy(src, file_system="flash:") + + logged = "\n".join(captured.output) + self.assertIn("Incorrect Login/Password", logged) + self.assertNotIn("p@ss/w0rd", logged) + self.assertNotIn("p%40ss%2Fw0rd", logged) + self.assertNotIn("p@ss/w0rd", str(raised.exception)) + self.assertNotIn("p%40ss%2Fw0rd", str(raised.exception)) + self.assertEqual( + registered, + {"password": "pass", "file_copy_token": "p@ss/w0rd", "file_copy_token_encoded": "p%40ss%2Fw0rd"}, + ) + self.assertEqual(no_log, {"password": "pass"}) + @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."""