Source code for jnpr.junos.utils.scp

from __future__ import absolute_import
import inspect

from scp import SCPClient
from jnpr.junos.utils.ssh_client import open_ssh_client

Secure Copy Utility

[docs]class SCP(object): """ The SCP utility is used to conjunction with :class:`jnpr.junos.utils.sw.SW` when transferring the Junos image to the device. The :class:`SCP` can be used for other secure-copy use-cases as well; it is implemented to support the python *context-manager* pattern. For example:: from jnpr.junos.utils.scp import SCP with SCP(dev, progress=True) as scp: scp.put(package, remote_path) """
[docs] def __init__(self, junos, **scpargs): """ Constructor that wraps :py:mod:`paramiko` and :py:mod:`scp` objects. :param Device junos: the Device object :param kvargs scpargs: any additional args to be passed to paramiko SCP """ self._junos = junos if self._junos.__dict__.get("_mode") is not None: raise RuntimeError("SCP is not supported with Console mode") self._scpargs = scpargs self._by10pct = 0 self._user_progress = self._scpargs.get("progress") self._ssh = None if self._user_progress is True: self._scpargs["progress"] = self._scp_progress elif callable(self._user_progress): # User case also define progress with 3 params, the way scp module # expects. Function will take path, total size, transferred. # spec = inspect.getargspec(self._user_progress) if (len(spec.args) == 3 and spec.args[0] != "self") or ( len(spec.args) == 4 and spec.args[0] == "self" ): self._scpargs["progress"] = self._user_progress else: # this will override the function _progress defined for this # class to use progress provided by user. self._progress = lambda report: self._user_progress(self._junos, report) self._scpargs["progress"] = self._scp_progress
def _progress(self, report): """ simple progress report function """ print(self._junos.hostname + ": " + report) def _scp_progress(self, _path, _total, _xfrd): # calculate current percentage xferd pct = int(float(_xfrd) / float(_total) * 100) # if 10% more has been copied, then print a message if 0 == (pct % 10) and pct != self._by10pct: self._by10pct = pct self._progress("%s: %s / %s (%s%%)" % (_path, _xfrd, _total, str(pct)))
[docs] def open(self, **scpargs): """ Creates an instance of the scp object and return to caller for use. .. note:: This method uses the same username/password authentication credentials as used by :class:`jnpr.junos.device.Device`. It can also use ``ssh_private_key_file`` option if provided to the :class:`jnpr.junos.device.Device` :returns: SCPClient object """ # @@@ should check for multi-calls to connect to ensure we don't keep # @@@ opening new connections self._ssh = open_ssh_client(dev=self._junos) return SCPClient(self._ssh.get_transport(), **scpargs)
[docs] def close(self): """ Closes the ssh/scp connection to the device """ self._ssh.close()
# ------------------------------------------------------------------------- # CONTEXT MANAGER # ------------------------------------------------------------------------- def __enter__(self): return**self._scpargs) def __exit__(self, exc_ty, exc_val, exc_tb): self.close()