52347

Build # 52347

Build Type

push

coveralls-python

Committed by Red_M

Commit Message

Fix bug with pty argument to open a single use exec command channel. Add test for `redssh.RedSSH().execute_command()`. Version bump.

Run Details

9 of 9 new or added lines in 4 files covered. (100.0%)

323 of 1605 relevant lines covered (20.12%)

0.2 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

44.93

/redssh/redssh.py

# RedSSH
# Copyright (C) 2018 - 2022 Red_M ( http://bitbucket.com/Red_M )

# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.

# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

import os
import re
import time
import socket

from . import exceptions
from . import enums
from . import clients
from . import utils


class RedSSH(object):
    '''
    Instances the start of an SSH connection.
    Extra options are available after :func:`redssh.RedSSH.connect` is called.

    :param encoding: Set the encoding to something other than the default of ``'utf8'`` when your target SSH server doesn't return UTF-8.
    :type encoding: ``str``
    '''
    def __init__(self,encoding='utf8',terminal='vt100',known_hosts=None,ssh_host_key_verification=enums.SSHHostKeyVerify.warn,
        ssh_keepalive_interval=0.0,set_flags={},method_preferences={},callbacks={},auto_terminate_tunnels=False,tcp_nodelay=False):
        self.debug = False
        self.client = self.pick_client()(encoding=encoding,terminal=terminal,known_hosts=known_hosts,ssh_host_key_verification=ssh_host_key_verification,
            ssh_keepalive_interval=ssh_keepalive_interval,set_flags=set_flags,method_preferences=method_preferences,callbacks=callbacks,
            auto_terminate_tunnels=auto_terminate_tunnels,tcp_nodelay=tcp_nodelay)
        self.enums = self.client.enums

    def pick_client(self,ssh_client=None,custom_ssh_clients={}):
        default_client = clients.default_client
        if ssh_client==None:
            ssh_client = default_client
        client_pool = {}
        client_pool.update(clients.enabled_clients)
        client_pool.update(custom_ssh_clients)
        return(client_pool.get(ssh_client,clients.enabled_clients[default_client]))

    def eof(self):
        '''
        Returns ``True`` or ``False`` when the main channel has recieved an ``EOF``.
        '''
        return(self.client.eof())

    def methods(self, method):
        '''
        Returns what value was settled on during session negotiation.
        '''
        return(self.client.methods(method))

    def setenv(self, varname, value):
        '''
        Set an environment variable on the channel.

        :param varname: Name of environment variable to set on the remote channel.
        :type varname: ``str``
        :param value: Value to set ``varname`` to.
        :type value: ``str``
        :return: ``None``
        '''
        self.client.setenv(varname, value)

    def connect(self,hostname,port=22,username='',password=None,
        allow_agent=False,host_based=None,key_filepath=None,passphrase=None,
        look_for_keys=False,sock=None,timeout=None):
        '''
        .. warning::
            Some authentication methods are not yet supported!

        :param hostname: Hostname to connect to.
        :type hostname: ``str``
        :param port: SSH port to connect to.
        :type port: ``int``
        :param username: Username to connect as to the remote server.
        :type username: ``str``
        :param password: Password to offer to the remote server for authentication.
        :type password: ``str``
        :param allow_agent: Allow the local SSH key agent to offer the keys held in it for authentication.
        :type allow_agent: ``bool``
        :param host_based: Allow the local SSH host keys to be used for authentication. NOT IMPLEMENTED!
        :type host_based: ``bool``
        :param key_filepath: Array of filenames to offer to the remote server. Can be a string for a single key.
        :type key_filepath: ``array``/``str``
        :param passphrase: Passphrase to decrypt any keys offered to the remote server.
        :type passphrase: ``str``
        :param look_for_keys: Enable offering keys in ``~/.ssh`` automatically. NOT IMPLEMENTED!
        :type look_for_keys: ``bool``
        :param sock: A pre-connected socket to the remote server. Useful if you have strange network requirements.
        :type sock: :func:`socket.socket`
        :param timeout: Timeout for the socket connection to the remote server.
        :type timeout: ``float``
        '''
        self.client.before_connect_options()
        self.client.connect(hostname,port,username,password,allow_agent,host_based,key_filepath,passphrase,look_for_keys,sock,timeout)
        self.client.after_connect_options()

    def read(self,block=False):
        '''
        Recieve data from the remote session.
        Only works if the current session has made it past the login process.

        :param block: Block until data is received from the remote server. ``True``
            will block until data is recieved and ``False`` may return ``b''`` if no data is available from the remote server.
        :type block: ``bool``
        :return: ``generator`` - A generator of byte strings that has been recieved in the time given.
        '''
        return(self.client.read(block))

    def send(self,string):
        '''
        Send data to the remote session.
        Only works if the current session has made it past the login process.

        :param string: String to send to the remote session.
        :type string: ``str``
        :return: ``int`` - Amount of bytes sent to remote machine.
        '''
        return(self.client.send(string))

    def write(self,string):
        '''
        See :func:`redssh.RedSSH.send`
        '''
        return(self.send(string))

    def flush(self):
        '''
        Flush all data on the primary channel's stdin to the remote connection.
        Only works if connected, otherwise returns ``0``.

        :return: ``int`` - Amount of bytes sent to remote machine.
        '''
        if self.client.past_login==True:
            return(self.client.flush())
        return(0)

    def last_error(self):
        '''
        Get the last error from the client session.

        :return: ``str``
        '''
        return(self.client.last_error())

    def execute_command(self,command,env=None,channel=None,pty=False):
        '''
        Run a command. This will block as the command executes.

        :param command: Command to execute.
        :type command: ``str``
        :param env: Environment variables to set during ``command``.
        :type env: ``dict``
        :param channel: Use an existing SSH channel instead of spawning a new one.
        :type channel: ``redssh.RedSSH.channel``
        :param pty: Request a pty for the command to be executed via.
        :type pty: ``bool``
        :return: ``tuple (int, str)`` - of ``(return_code, command_output)``
        '''
        return(self.client.execute_command(command,env=env,channel=channel,pty=pty))

    def start_sftp(self):
        '''
        Start the SFTP client.
        The client will be available at `self.sftp` and will be an instance of `redssh.sftp.RedSFTP`

        :return: ``None``
        '''
        self.client.start_sftp()
        if self.client.past_login and self.client.__check_for_attr__('sftp')==True:
            self.sftp = utils.ObjectProxy(self.client,'sftp')

    def start_scp(self):
        '''
        Start the SCP client.
        If the client or server doesn't support SCP, SFTP will be started instead, this is due to SCP being deprecated.

        :return: ``None``
        '''
        self.client.start_scp()
        if self.client.past_login and self.client.__check_for_attr__('scp')==True:
            self.scp = utils.ObjectProxy(self.client,'scp')


    def forward_x11(self):
        '''
        Start forwarding an X11 display.

        :return: ``None``
        '''
        self.client.forward_x11()

    def local_tunnel(self,local_port,remote_host,remote_port,bind_addr='127.0.0.1',error_level=enums.TunnelErrorLevel.debug):
        '''

        Forwards a port on the remote machine the same way the ``-L`` option does for the OpenSSH client.

        Providing a ``0`` for the local port will mean the OS will assign an unbound port for you.
        This port number will be provided to you by this function.

        :param local_port: The local port on the local machine to bind to.
        :type local_port: ``int``
        :param remote_host: The remote host to connect to via the remote machine.
        :type remote_host: ``str``
        :param remote_port: The remote host's port to connect to via the remote machine.
        :type remote_port: ``int``
        :param bind_addr: The bind address on this machine to bind to for the local port.
        :type bind_addr: ``str``
        :param error_level: The level of verbosity that errors in tunnel threads will use.
        :type error_level: :class:`redssh.enums.TunnelErrorLevel`
        :return: ``int`` The local port that has been bound.
        '''
        return(self.client.local_tunnel(local_port,remote_host,remote_port,bind_addr,error_level))

    def remote_tunnel(self,local_port,remote_host,remote_port,bind_addr='127.0.0.1',error_level=enums.TunnelErrorLevel.warn):
        '''

        Forwards a port to the remote machine via the local machine the same way the ``-R`` option does for the OpenSSH client.

        :param local_port: The local port on the remote side for clients to connect to.
        :type local_port: ``int``
        :param remote_host: The remote host to connect to via the local machine.
        :type remote_host: ``str``
        :param remote_port: The remote host's port to connect to via the local machine.
        :type remote_port: ``int``
        :param error_level: The level of verbosity that errors in tunnel threads will use.
        :type error_level: :class:`redssh.enums.TunnelErrorLevel`
        :return: ``None``
        '''
        return(self.client.remote_tunnel(local_port,remote_host,remote_port,bind_addr,error_level))

    def dynamic_tunnel(self,local_port,bind_addr='127.0.0.1',error_level=enums.TunnelErrorLevel.warn):
        '''

        Opens a SOCKS proxy AKA gateway or dynamic port the same way the ``-D`` option does for the OpenSSH client.

        Providing a ``0`` for the local port will mean the OS will assign an unbound port for you.
        This port number will be provided to you by this function.

        :param local_port: The local port on the local machine to bind to.
        :type local_port: ``int``
        :param bind_addr: The bind address on this machine to bind to for the local port.
        :type bind_addr: ``str``
        :param error_level: The level of verbosity that errors in tunnel threads will use.
        :type error_level: :class:`redssh.enums.TunnelErrorLevel`
        :return: ``int`` The local port that has been bound.
        '''
        return(self.client.dynamic_tunnel(local_port,bind_addr,error_level))

    def tunnel_is_alive(self,tunnel_type,sport,rhost=None,rport=None,bind_addr='127.0.0.1'):
        '''

        Checks if a tunnel is alive.
        Provide the same arguments to this that was given for openning the tunnel.

        Examples:

        `local_tunnel(9999,'localhost',8888)` would be `tunnel_is_alive(redssh.enums.TunnelType.local,9999,'localhost',8888)`

        `remote_tunnel(7777,'localhost',8888)` would be `tunnel_is_alive(redssh.enums.TunnelType.remote,7777,'localhost',8888)`

        `dynamic_tunnel(9999)` would be `tunnel_is_alive(redssh.enums.TunnelType.dynamic,9999)`

        `dynamic_tunnel(9999,'10.0.0.1')` would be `tunnel_is_alive(redssh.enums.TunnelType.dynamic,9999,bind_addr='10.0.0.1')`

        :param tunnel_type: The tunnel type to shutdown.
        :type tunnel_type: :class:`redssh.enums.TunnelType`
        :param sport: The bound port for local and dynamic tunnels or the local port on the remote side for remote tunnels.
        :type sport: ``str``
        :param rhost: The remote host for local and remote tunnels.
        :type rhost: ``str``
        :param rport: The remote port for local and remote tunnels.
        :type rport: ``int``
        :param bind_addr: The bind address used for local and dynamic tunnels.
        :type bind_addr: ``str``
        :return: ``bool``, if bad tunnel type provided returns ``None``
        '''
        return(self.client.tunnel_is_alive(tunnel_type,sport,rhost,rport,bind_addr))

    def shutdown_tunnel(self,tunnel_type,sport,rhost=None,rport=None,bind_addr='127.0.0.1'):
        '''

        Closes an open tunnel.
        Provide the same arguments to this that was given for openning the tunnel.

        Examples:

        `local_tunnel(9999,'localhost',8888)` would be `shutdown_tunnel(redssh.enums.TunnelType.local,9999,'localhost',8888)`

        `remote_tunnel(7777,'localhost',8888)` would be `shutdown_tunnel(redssh.enums.TunnelType.remote,7777,'localhost',8888)`

        `dynamic_tunnel(9999)` would be `shutdown_tunnel(redssh.enums.TunnelType.dynamic,9999)`

        `dynamic_tunnel(9999,'10.0.0.1')` would be `shutdown_tunnel(redssh.enums.TunnelType.dynamic,9999,bind_addr='10.0.0.1')`

        :param tunnel_type: The tunnel type to shutdown.
        :type tunnel_type: :class:`redssh.enums.TunnelType`
        :param sport: The bound port for local and dynamic tunnels or the local port on the remote side for remote tunnels.
        :type sport: ``str``
        :param rhost: The remote host for local and remote tunnels.
        :type rhost: ``str``
        :param rport: The remote port for local and remote tunnels.
        :type rport: ``int``
        :param bind_addr: The bind address used for local and dynamic tunnels.
        :type bind_addr: ``str``
        :return: ``None``
        '''
        self.client.shutdown_tunnel(tunnel_type,sport,rhost,rport,bind_addr)


    def close_tunnels(self):
        '''
        Closes all SSH tunnels if any are open.
        '''
        self.client.close_tunnels()

    def exit(self):
        '''
        Kill the current session if connected.
        '''
        self.client.exit()

1	# RedSSH
2	# Copyright (C) 2018 - 2022 Red_M ( http://bitbucket.com/Red_M )
3
4	# This program is free software; you can redistribute it and/or modify
5	# it under the terms of the GNU General Public License as published by
6	# the Free Software Foundation; either version 2 of the License, or
7	# (at your option) any later version.
8
9	# This program is distributed in the hope that it will be useful,
10	# but WITHOUT ANY WARRANTY; without even the implied warranty of
11	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12	# GNU General Public License for more details.
13
14	# You should have received a copy of the GNU General Public License along
15	# with this program; if not, write to the Free Software Foundation, Inc.,
16	# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17
18	import os	1✔
19	import re	1✔
20	import time	1✔
21	import socket	1✔
22
23	from . import exceptions	1✔
24	from . import enums	1✔
25	from . import clients	1✔
26	from . import utils	1✔
27
28
29	class RedSSH(object):	1✔
30	'''
31	Instances the start of an SSH connection.
32	Extra options are available after :func:`redssh.RedSSH.connect` is called.
33
34	:param encoding: Set the encoding to something other than the default of ``'utf8'`` when your target SSH server doesn't return UTF-8.
35	:type encoding: ``str``
36	'''
37	def __init__(self,encoding='utf8',terminal='vt100',known_hosts=None,ssh_host_key_verification=enums.SSHHostKeyVerify.warn,	1✔
38	ssh_keepalive_interval=0.0,set_flags={},method_preferences={},callbacks={},auto_terminate_tunnels=False,tcp_nodelay=False):
39	self.debug = False	×
40	self.client = self.pick_client()(encoding=encoding,terminal=terminal,known_hosts=known_hosts,ssh_host_key_verification=ssh_host_key_verification,	×
41	ssh_keepalive_interval=ssh_keepalive_interval,set_flags=set_flags,method_preferences=method_preferences,callbacks=callbacks,
42	auto_terminate_tunnels=auto_terminate_tunnels,tcp_nodelay=tcp_nodelay)
43	self.enums = self.client.enums	×
44
45	def pick_client(self,ssh_client=None,custom_ssh_clients={}):	1✔
46	default_client = clients.default_client	×
47	if ssh_client==None:	×
48	ssh_client = default_client	×
49	client_pool = {}	×
50	client_pool.update(clients.enabled_clients)	×
51	client_pool.update(custom_ssh_clients)	×
52	return(client_pool.get(ssh_client,clients.enabled_clients[default_client]))	×
53
54	def eof(self):	1✔
55	'''
56	Returns ``True`` or ``False`` when the main channel has recieved an ``EOF``.
57	'''
58	return(self.client.eof())	×
59
60	def methods(self, method):	1✔
61	'''
62	Returns what value was settled on during session negotiation.
63	'''
64	return(self.client.methods(method))	×
65
66	def setenv(self, varname, value):	1✔
67	'''
68	Set an environment variable on the channel.
69
70	:param varname: Name of environment variable to set on the remote channel.
71	:type varname: ``str``
72	:param value: Value to set ``varname`` to.
73	:type value: ``str``
74	:return: ``None``
75	'''
76	self.client.setenv(varname, value)	×
77
78	def connect(self,hostname,port=22,username='',password=None,	1✔
79	allow_agent=False,host_based=None,key_filepath=None,passphrase=None,
80	look_for_keys=False,sock=None,timeout=None):
81	'''
82	.. warning::
83	Some authentication methods are not yet supported!
84
85	:param hostname: Hostname to connect to.
86	:type hostname: ``str``
87	:param port: SSH port to connect to.
88	:type port: ``int``
89	:param username: Username to connect as to the remote server.
90	:type username: ``str``
91	:param password: Password to offer to the remote server for authentication.
92	:type password: ``str``
93	:param allow_agent: Allow the local SSH key agent to offer the keys held in it for authentication.
94	:type allow_agent: ``bool``
95	:param host_based: Allow the local SSH host keys to be used for authentication. NOT IMPLEMENTED!
96	:type host_based: ``bool``
97	:param key_filepath: Array of filenames to offer to the remote server. Can be a string for a single key.
98	:type key_filepath: ``array``/``str``
99	:param passphrase: Passphrase to decrypt any keys offered to the remote server.
100	:type passphrase: ``str``
101	:param look_for_keys: Enable offering keys in ``~/.ssh`` automatically. NOT IMPLEMENTED!
102	:type look_for_keys: ``bool``
103	:param sock: A pre-connected socket to the remote server. Useful if you have strange network requirements.
104	:type sock: :func:`socket.socket`
105	:param timeout: Timeout for the socket connection to the remote server.
106	:type timeout: ``float``
107	'''
108	self.client.before_connect_options()	×
109	self.client.connect(hostname,port,username,password,allow_agent,host_based,key_filepath,passphrase,look_for_keys,sock,timeout)	×
110	self.client.after_connect_options()	×
111
112	def read(self,block=False):	1✔
113	'''
114	Recieve data from the remote session.
115	Only works if the current session has made it past the login process.
116
117	:param block: Block until data is received from the remote server. ``True``
118	will block until data is recieved and ``False`` may return ``b''`` if no data is available from the remote server.
119	:type block: ``bool``
120	:return: ``generator`` - A generator of byte strings that has been recieved in the time given.
121	'''
122	return(self.client.read(block))	×
123
124	def send(self,string):	1✔
125	'''
126	Send data to the remote session.
127	Only works if the current session has made it past the login process.
128
129	:param string: String to send to the remote session.
130	:type string: ``str``
131	:return: ``int`` - Amount of bytes sent to remote machine.
132	'''
133	return(self.client.send(string))	×
134
135	def write(self,string):	1✔
136	'''
137	See :func:`redssh.RedSSH.send`
138	'''
139	return(self.send(string))	×
140
141	def flush(self):	1✔
142	'''
143	Flush all data on the primary channel's stdin to the remote connection.
144	Only works if connected, otherwise returns ``0``.
145
146	:return: ``int`` - Amount of bytes sent to remote machine.
147	'''
148	if self.client.past_login==True:	×
149	return(self.client.flush())	×
150	return(0)	×
151
152	def last_error(self):	1✔
153	'''
154	Get the last error from the client session.
155
156	:return: ``str``
157	'''
158	return(self.client.last_error())	×
159
160	def execute_command(self,command,env=None,channel=None,pty=False):	1✔
161	'''
162	Run a command. This will block as the command executes.
163
164	:param command: Command to execute.
165	:type command: ``str``
166	:param env: Environment variables to set during ``command``.
167	:type env: ``dict``
168	:param channel: Use an existing SSH channel instead of spawning a new one.
169	:type channel: ``redssh.RedSSH.channel``
170	:param pty: Request a pty for the command to be executed via.
171	:type pty: ``bool``
172	:return: ``tuple (int, str)`` - of ``(return_code, command_output)``
173	'''
174	return(self.client.execute_command(command,env=env,channel=channel,pty=pty))	×
175
176	def start_sftp(self):	1✔
177	'''
178	Start the SFTP client.
179	The client will be available at `self.sftp` and will be an instance of `redssh.sftp.RedSFTP`
180
181	:return: ``None``
182	'''
183	self.client.start_sftp()	×
184	if self.client.past_login and self.client.__check_for_attr__('sftp')==True:	×
185	self.sftp = utils.ObjectProxy(self.client,'sftp')	×
186
187	def start_scp(self):	1✔
188	'''
189	Start the SCP client.
190	If the client or server doesn't support SCP, SFTP will be started instead, this is due to SCP being deprecated.
191
192	:return: ``None``
193	'''
194	self.client.start_scp()	×
195	if self.client.past_login and self.client.__check_for_attr__('scp')==True:	×
196	self.scp = utils.ObjectProxy(self.client,'scp')	×
197
198
199	def forward_x11(self):	1✔
200	'''
201	Start forwarding an X11 display.
202
203	:return: ``None``
204	'''
205	self.client.forward_x11()	×
206
207	def local_tunnel(self,local_port,remote_host,remote_port,bind_addr='127.0.0.1',error_level=enums.TunnelErrorLevel.debug):	1✔
208	'''
209
210	Forwards a port on the remote machine the same way the ``-L`` option does for the OpenSSH client.
211
212	Providing a ``0`` for the local port will mean the OS will assign an unbound port for you.
213	This port number will be provided to you by this function.
214
215	:param local_port: The local port on the local machine to bind to.
216	:type local_port: ``int``
217	:param remote_host: The remote host to connect to via the remote machine.
218	:type remote_host: ``str``
219	:param remote_port: The remote host's port to connect to via the remote machine.
220	:type remote_port: ``int``
221	:param bind_addr: The bind address on this machine to bind to for the local port.
222	:type bind_addr: ``str``
223	:param error_level: The level of verbosity that errors in tunnel threads will use.
224	:type error_level: :class:`redssh.enums.TunnelErrorLevel`
225	:return: ``int`` The local port that has been bound.
226	'''
227	return(self.client.local_tunnel(local_port,remote_host,remote_port,bind_addr,error_level))	×
228
229	def remote_tunnel(self,local_port,remote_host,remote_port,bind_addr='127.0.0.1',error_level=enums.TunnelErrorLevel.warn):	1✔
230	'''
231
232	Forwards a port to the remote machine via the local machine the same way the ``-R`` option does for the OpenSSH client.
233
234	:param local_port: The local port on the remote side for clients to connect to.
235	:type local_port: ``int``
236	:param remote_host: The remote host to connect to via the local machine.
237	:type remote_host: ``str``
238	:param remote_port: The remote host's port to connect to via the local machine.
239	:type remote_port: ``int``
240	:param error_level: The level of verbosity that errors in tunnel threads will use.
241	:type error_level: :class:`redssh.enums.TunnelErrorLevel`
242	:return: ``None``
243	'''
244	return(self.client.remote_tunnel(local_port,remote_host,remote_port,bind_addr,error_level))	×
245
246	def dynamic_tunnel(self,local_port,bind_addr='127.0.0.1',error_level=enums.TunnelErrorLevel.warn):	1✔
247	'''
248
249	Opens a SOCKS proxy AKA gateway or dynamic port the same way the ``-D`` option does for the OpenSSH client.
250
251	Providing a ``0`` for the local port will mean the OS will assign an unbound port for you.
252	This port number will be provided to you by this function.
253
254	:param local_port: The local port on the local machine to bind to.
255	:type local_port: ``int``
256	:param bind_addr: The bind address on this machine to bind to for the local port.
257	:type bind_addr: ``str``
258	:param error_level: The level of verbosity that errors in tunnel threads will use.
259	:type error_level: :class:`redssh.enums.TunnelErrorLevel`
260	:return: ``int`` The local port that has been bound.
261	'''
262	return(self.client.dynamic_tunnel(local_port,bind_addr,error_level))	×
263
264	def tunnel_is_alive(self,tunnel_type,sport,rhost=None,rport=None,bind_addr='127.0.0.1'):	1✔
265	'''
266
267	Checks if a tunnel is alive.
268	Provide the same arguments to this that was given for openning the tunnel.
269
270	Examples:
271
272	`local_tunnel(9999,'localhost',8888)` would be `tunnel_is_alive(redssh.enums.TunnelType.local,9999,'localhost',8888)`
273
274	`remote_tunnel(7777,'localhost',8888)` would be `tunnel_is_alive(redssh.enums.TunnelType.remote,7777,'localhost',8888)`
275
276	`dynamic_tunnel(9999)` would be `tunnel_is_alive(redssh.enums.TunnelType.dynamic,9999)`
277
278	`dynamic_tunnel(9999,'10.0.0.1')` would be `tunnel_is_alive(redssh.enums.TunnelType.dynamic,9999,bind_addr='10.0.0.1')`
279
280	:param tunnel_type: The tunnel type to shutdown.
281	:type tunnel_type: :class:`redssh.enums.TunnelType`
282	:param sport: The bound port for local and dynamic tunnels or the local port on the remote side for remote tunnels.
283	:type sport: ``str``
284	:param rhost: The remote host for local and remote tunnels.
285	:type rhost: ``str``
286	:param rport: The remote port for local and remote tunnels.
287	:type rport: ``int``
288	:param bind_addr: The bind address used for local and dynamic tunnels.
289	:type bind_addr: ``str``
290	:return: ``bool``, if bad tunnel type provided returns ``None``
291	'''
292	return(self.client.tunnel_is_alive(tunnel_type,sport,rhost,rport,bind_addr))	×
293
294	def shutdown_tunnel(self,tunnel_type,sport,rhost=None,rport=None,bind_addr='127.0.0.1'):	1✔
295	'''
296
297	Closes an open tunnel.
298	Provide the same arguments to this that was given for openning the tunnel.
299
300	Examples:
301
302	`local_tunnel(9999,'localhost',8888)` would be `shutdown_tunnel(redssh.enums.TunnelType.local,9999,'localhost',8888)`
303
304	`remote_tunnel(7777,'localhost',8888)` would be `shutdown_tunnel(redssh.enums.TunnelType.remote,7777,'localhost',8888)`
305
306	`dynamic_tunnel(9999)` would be `shutdown_tunnel(redssh.enums.TunnelType.dynamic,9999)`
307
308	`dynamic_tunnel(9999,'10.0.0.1')` would be `shutdown_tunnel(redssh.enums.TunnelType.dynamic,9999,bind_addr='10.0.0.1')`
309
310	:param tunnel_type: The tunnel type to shutdown.
311	:type tunnel_type: :class:`redssh.enums.TunnelType`
312	:param sport: The bound port for local and dynamic tunnels or the local port on the remote side for remote tunnels.
313	:type sport: ``str``
314	:param rhost: The remote host for local and remote tunnels.
315	:type rhost: ``str``
316	:param rport: The remote port for local and remote tunnels.
317	:type rport: ``int``
318	:param bind_addr: The bind address used for local and dynamic tunnels.
319	:type bind_addr: ``str``
320	:return: ``None``
321	'''
322	self.client.shutdown_tunnel(tunnel_type,sport,rhost,rport,bind_addr)	×
323
324
325	def close_tunnels(self):	1✔
326	'''
327	Closes all SSH tunnels if any are open.
328	'''
329	self.client.close_tunnels()	×
330
331	def exit(self):	1✔
332	'''
333	Kill the current session if connected.
334	'''
335	self.client.exit()	×

Red-M / RedSSH / 52347

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous