Source code for jnpr.junos.utils.start_shell
import paramiko
from select import select
import re
_JUNOS_PROMPT = '> '
_SHELL_PROMPT = '(%|#)\s'
_SELECT_WAIT = 0.1
_RECVSZ = 1024
[docs]class StartShell(object):
"""
Junos shell execution utility. This utility is written to
support the "context manager" design pattern. For example::
def _ssh_exec(self, command):
with StartShell(self._dev) as sh:
got = sh.run(command)
return got
"""
[docs] def __init__(self, nc):
"""
Utility Constructor
:param Device nc: The Device object
"""
self._nc = nc
[docs] def wait_for(self, this=_SHELL_PROMPT):
"""
Wait for the result of the command, expecting **this** prompt.
:param str this: expected string/pattern.
:returns: resulting string of data in a list
:rtype: list
.. warning:: need to add a timeout safeguard
"""
chan = self._chan
got = []
while True:
rd, wr, err = select([chan], [], [], _SELECT_WAIT)
if rd:
data = chan.recv(_RECVSZ)
if isinstance(data, bytes):
data = data.decode('utf-8')
got.append(data)
if re.search(r'{0}\s?$'.format(this), data):
break
return got
[docs] def send(self, data):
"""
Send the command **data** followed by a newline character.
:param str data: the data to write out onto the shell.
:returns: result of SSH channel send
"""
self._chan.send(data)
self._chan.send('\n')
[docs] def open(self):
"""
Open an ssh-client connection and issue the 'start shell' command to
drop into the Junos shell (csh). This process opens a
:class:`paramiko.SSHClient` instance.
"""
junos = self._nc
client = paramiko.SSHClient()
client.load_system_host_keys()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(hostname=junos.hostname,
port=(22, junos._port)[junos.hostname == 'localhost'],
username=junos._auth_user,
password=junos._auth_password,
)
chan = client.invoke_shell()
self._client = client
self._chan = chan
got = self.wait_for(r'(%|>|#)')
if got[-1].endswith(_JUNOS_PROMPT):
self.send('start shell')
self.wait_for(_SHELL_PROMPT)
[docs] def close(self):
""" Close the SSH client channel """
self._chan.close()
self._client.close()
[docs] def run(self, command, this=_SHELL_PROMPT):
"""
Run a shell command and wait for the response. The return is a
tuple. The first item is True/False if exit-code is 0. The second
item is the output of the command.
:param str command: the shell command to execute
:param str this: the exected shell-prompt to wait for
:returns: (last_ok, result of the executed shell command (str) )
.. note:: as a *side-effect* this method will set the ``self.last_ok``
property. This property is set to ``True`` if ``$?`` is
"0"; indicating the last shell command was successful.
"""
# run the command and capture the output
self.send(command)
got = ''.join(self.wait_for(this))
# use $? to get the exit code of the command
self.send('echo $?')
rc = ''.join(self.wait_for(this))
self.last_ok = True if rc.find('0') > 0 else False
return (self.last_ok, got)
# -------------------------------------------------------------------------
# CONTEXT MANAGER
# -------------------------------------------------------------------------
def __enter__(self):
self.open()
return self
def __exit__(self, exc_ty, exc_val, exc_tb):
self.close()