Add Arista EOS remote file copy#365
Conversation
8ec398a to
d34fe08
Compare
44f96c6 to
6a32a97
Compare
6a32a97 to
3fc3736
Compare
|
@jeffkala, I appreciate the comments. I've applied your suggestions. |
jeffkala
left a comment
There was a problem hiding this comment.
I'm good with it, thanks for updating my first set of reviews.
| """Get the checksum of a remote file on Arista EOS device using netmiko SSH. | ||
|
|
||
| Uses Arista's 'verify' command via SSH to compute file checksums. | ||
| Note, Netmiko FileTransfer only supports `verify /md5` |
There was a problem hiding this comment.
I was hoping the netmiko FileTransfer would solve this problem, but there is an issue where it is limited to md5.
https://ktbyers.github.io/netmiko/docs/netmiko/arista/index.html > Expand Source Code
def remote_md5(
self, base_cmd: str = "verify /md5", remote_file: Optional[str] = None
) -> str:
if remote_file is None:
if self.direction == "put":
remote_file = self.dest_file
elif self.direction == "get":
remote_file = self.source_file
remote_md5_cmd = f"{base_cmd} file:{self.file_system}/{remote_file}"
dest_md5 = self.ssh_ctl_chan._send_command_str(remote_md5_cmd, read_timeout=600)
dest_md5 = self.process_md5(dest_md5)
return dest_md5
|
|
||
| # Parse the checksum from the output | ||
| # Expected format: verify /sha512 (flash:nautobot.png) = <checksum> | ||
| match = re.search(r"=\s*([a-fA-F0-9]+)", result) |
There was a problem hiding this comment.
I don't like how imprecise this regex is. Can we just do if remote_checksum.lower() in result.lower():?
|
|
||
| # Use Arista's verify command to get the checksum | ||
| # Example: verify /sha512 flash:nautobot.png | ||
| command = f"verify /{hashing_algorithm} {path}" |
There was a problem hiding this comment.
We should handle invalid hashing_algorithm values and raise the correct exception so the user knows that the checksum type is invalid instead of just saying that it failed to parse. If they select SHA in their SoftwareImageFile they may never understand that their checksum type is wrong and will just assume that the file transfer is failing since % Invalid input is the only message they'll see:
Arista7050TX-64-R#verify /asdf flash:/EOS-4.15.4F.swi
% Invalid input
|
|
||
| def _build_url_copy_command_simple(self, src, file_system): | ||
| """Build copy command for simple URL-based transfers (TFTP, HTTP, HTTPS without credentials).""" | ||
| return f"copy {src.download_url} {file_system}", False |
There was a problem hiding this comment.
This should include the filename in the destination or it will default to the filename in the url which may not be correct. For example the filename in sftpgo's web server defaults to download because the url is http://10.1.100.220:8080/web/client/pubshares/QVuvQMzugSegd2e7krDwr7/download
|
|
||
| def _build_url_copy_command_with_creds(self, src, file_system): | ||
| """Build copy command for URL-based transfers with credentials (HTTP/HTTPS/SCP/FTP/SFTP).""" | ||
| parsed = urlparse(src.download_url) |
There was a problem hiding this comment.
We probably need to fail if parsed.query is not blank since a url with a query contains a ? which doesn't work in the EOS command line.
| raise ValueError(f"Unsupported scheme: {src.scheme}") | ||
|
|
||
| # Build command based on scheme and credentials | ||
| command_builders = { |
There was a problem hiding this comment.
If include_username is not required we would probably want to do something like this:
if src.scheme == "tftp" or src.username is None:
command, detect_prompt = self._build_url_copy_command_simple(src, file_system)
else:
command, detect_prompt = self._build_url_copy_command_with_creds(src, file_system)If they didn't provide credentials and the copy fails, we'll bubble up that failure to the user.
There was a problem hiding this comment.
May also need to account for src.username is None and src.token is not None for URLs like https://token@example.com/file.bin
| device_checksum = ( | ||
| self.get_remote_checksum(filename, hashing_algorithm=hashing_algorithm, **kwargs) if exists else None | ||
| ) | ||
| if checksum == device_checksum: |
There was a problem hiding this comment.
| if checksum == device_checksum: | |
| if checksum.lower() == device_checksum.lower(): |
| Returns: | ||
| (bool): True if the file is verified successfully, False otherwise. | ||
| """ | ||
| exists = self.check_file_exists(filename, **kwargs) |
There was a problem hiding this comment.
| exists = self.check_file_exists(filename, **kwargs) | |
| if not self.check_file_exists(filename, **kwargs): | |
| log.debug("...") # File not found log message here | |
| return False |
| with pytest.raises(TypeError) as exc_info: | ||
| device.remote_file_copy(src) | ||
|
|
||
| assert "src must be an instance of FileCopyModel" in str(exc_info.value) |
There was a problem hiding this comment.
99% of the apps ecosystem is unittest and self.assertRaises. Lets switch this over to those patterns for consistency.
| try: | ||
| from hypothesis import given | ||
| from hypothesis import strategies as st | ||
| except ImportError: | ||
| # Create dummy decorators if hypothesis is not available | ||
| def given(*args, **kwargs): | ||
| def decorator(func): | ||
| return func | ||
|
|
||
| return decorator | ||
|
|
||
| class _ST: | ||
| @staticmethod | ||
| def just(value): | ||
| return value | ||
|
|
||
| @staticmethod | ||
| def one_of(*args): | ||
| return args[0] | ||
|
|
||
| st = _ST() |
There was a problem hiding this comment.
We should stick with patterns that the rest of the team maintaining this is familiar with.
New Pull Request
Have you:
Change Notes
Added the remote file copy feature to Arista EOS devices.
Added unittests for remote file copy on Arista EOS devices. (with help from Claude)
Justification
This allows an Arista device to run the copy command directly.