Source code for pyssh.shell
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import warnings
import ctypes
from . import api
from . import compat
from . import exceptions as exp
[docs]class Shell(object):
"""
Shell session.
"""
_channel = None
def __init__(self, session, pty_size, env):
self.session_wrapper = session
self.session = session.session
self.pty_size = pty_size
self.env = env
@property
def channel(self):
if self._channel is not None:
return self._channel
self._channel = api.library.ssh_channel_new(self.session);
# Open ssh session
ret = api.library.ssh_channel_open_session(self._channel)
if ret != api.SSH_OK:
raise exp.ConnectionError("Error code: {0}".format(ret))
# Request pty
ret = api.library.ssh_channel_request_pty(self._channel)
if ret != api.SSH_OK:
raise exp.ConnectionError("Error code: {0}".format(ret))
# Request pty size
#ret = api.library.ssh_channel_request_pty_size(self._channel,
# self.pty_size[0], self.pty_size[1])
#if ret != api.SSH_OK:
# raise RuntimeError("Error code: {0}".format(ret))
# Request shell
ret = api.library.ssh_channel_request_shell(self._channel)
if ret != api.SSH_OK:
raise exp.ConnectionError("Error code: {0}".format(ret))
# Set environ variable if theese are available
if self.env:
for key, value in self.env.items():
_key, _value = key, value
if isinstance(_key, compat.text_type):
_key = compat.to_bytes(_key, encoding="utf-8")
if isinstance(_value, compat.text_type):
_value = compat.to_bytes(_value, encoding="utf-8")
res = api.library.ssh_channel_request_env(self.channel, _key, _value)
res = api.library.ssh_channel_request_shell(self.channel)
if res != api.SSH_OK:
msg = api.library.ssh_get_error(self.session)
print("Error: ", msg)
warnings.warn("Error on set {0} variable".format(key), RuntimeWarning)
return self._channel
[docs] def write(self, data):
"""
Write bytes to remote shell.
The `data` parameter accept both str and bytes, if you passes str (unicode) is
automatically converted to bytes using utf-8 encoding.
:param bytes data: arbitrary length of bytes.
:returns: a number of bytes written to remote shell.
:rtype: int
"""
if isinstance(data, compat.text_type):
data = compat.to_bytes(data, "utf-8")
written = api.library.ssh_channel_write(self.channel, data, len(data))
if written != len(data):
raise RuntimeError("Error on write")
return written
[docs] def read(self, n):
"""
Read bytes from remote shell.
This method always return value, if not bytes available to read
it returns an empty bytestring.
:param int n: number of bytes to read
:returns: bytestring of readed data.
:rtype: bytes
"""
res = api.library.ssh_channel_is_open(self.channel)
if res == 0:
raise RuntimeError("Channel is closed")
res = api.library.ssh_channel_is_eof(self.channel)
if res != 0:
return b""
buffer = ctypes.create_string_buffer(n)
readed = api.library.ssh_channel_read_nonblocking(self.channel, buffer, n, 0)
if readed < 0:
raise RuntimeError("Error on read")
return buffer.value
def __enter__(self):
return self
def __exit__(self, *args, **kwargs):
if self._channel is not None:
if api.library.ssh_channel_is_closed(self._channel) == 0:
api.library.ssh_channel_close(self._channel)
api.library.ssh_channel_send_eof(self._channel)
api.library.ssh_channel_free(self.channel)
self._channel = None