# Copyright (c) 2013-2024 by Ron Frederick <ronf@timeheart.net> and others.
#
# This program and the accompanying materials are made available under
# the terms of the Eclipse Public License v2.0 which accompanies this
# distribution and is available at:
#
# http://www.eclipse.org/legal/epl-2.0/
#
# This program may also be made available under the following secondary
# licenses when the conditions for such availability set forth in the
# Eclipse Public License v2.0 are satisfied:
#
# GNU General Public License, Version 2.0, or any later versions of
# that license
#
# SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
#
# Contributors:
# Ron Frederick - initial implementation, API, and documentation
"""SSH session handlers"""
from typing import TYPE_CHECKING, Any, AnyStr, Callable, Generic
from typing import Mapping, Optional, Tuple
if TYPE_CHECKING:
# pylint: disable=cyclic-import
from .channel import SSHClientChannel, SSHServerChannel
from .channel import SSHTCPChannel, SSHUNIXChannel, SSHTunTapChannel
DataType = Optional[int]
class SSHSession(Generic[AnyStr]):
"""SSH session handler"""
# pylint: disable=no-self-use,unused-argument
def connection_made(self, chan: Any) -> None:
"""Called when a channel is opened successfully"""
def connection_lost(self, exc: Optional[Exception]) -> None:
"""Called when a channel is closed
This method is called when a channel is closed. If the channel
is shut down cleanly, *exc* will be `None`. Otherwise, it
will be an exception explaining the reason for the channel close.
:param exc:
The exception which caused the channel to close, or
`None` if the channel closed cleanly.
:type exc: :class:`Exception`
"""
def session_started(self) -> None:
"""Called when the session is started
This method is called when a session has started up. For
client and server sessions, this will be called once a
shell, exec, or subsystem request has been successfully
completed. For TCP and UNIX domain socket sessions, it will
be called immediately after the connection is opened.
"""
def data_received(self, data: AnyStr, datatype: DataType) -> None:
"""Called when data is received on the channel
This method is called when data is received on the channel.
If an encoding was specified when the channel was created,
the data will be delivered as a string after decoding with
the requested encoding. Otherwise, the data will be delivered
as bytes.
:param data:
The data received on the channel
:param datatype:
The extended data type of the data, from :ref:`extended
data types <ExtendedDataTypes>`
:type data: `str` or `bytes`
"""
def eof_received(self) -> bool:
"""Called when EOF is received on the channel
This method is called when an end-of-file indication is received
on the channel, after which no more data will be received. If this
method returns `True`, the channel remains half open and data
may still be sent. Otherwise, the channel is automatically closed
after this method returns. This is the default behavior for
classes derived directly from :class:`SSHSession`, but not when
using the higher-level streams API. Because input is buffered
in that case, streaming sessions enable half-open channels to
allow applications to respond to input read after an end-of-file
indication is received.
"""
return False # pragma: no cover
def pause_writing(self) -> None:
"""Called when the write buffer becomes full
This method is called when the channel's write buffer becomes
full and no more data can be sent until the remote system
adjusts its window. While data can still be buffered locally,
applications may wish to stop producing new data until the
write buffer has drained.
"""
def resume_writing(self) -> None:
"""Called when the write buffer has sufficiently drained
This method is called when the channel's send window reopens
and enough data has drained from the write buffer to allow the
application to produce more data.
"""
[docs]
class SSHClientSession(SSHSession[AnyStr]):
"""SSH client session handler
Applications should subclass this when implementing an SSH client
session handler. The functions listed below should be implemented
to define application-specific behavior. In particular, the standard
`asyncio` protocol methods such as :meth:`connection_made`,
:meth:`connection_lost`, :meth:`data_received`, :meth:`eof_received`,
:meth:`pause_writing`, and :meth:`resume_writing` are all supported.
In addition, :meth:`session_started` is called as soon as the SSH
session is fully started, :meth:`xon_xoff_requested` can be used to
determine if the server wants the client to support XON/XOFF flow
control, and :meth:`exit_status_received` and
:meth:`exit_signal_received` can be used to receive session exit
information.
"""
# pylint: disable=no-self-use,unused-argument
[docs]
def connection_made(self, chan: 'SSHClientChannel[AnyStr]') -> None:
"""Called when a channel is opened successfully
This method is called when a channel is opened successfully. The
channel parameter should be stored if needed for later use.
:param chan:
The channel which was successfully opened.
:type chan: :class:`SSHClientChannel`
"""
[docs]
def xon_xoff_requested(self, client_can_do: bool) -> None:
"""XON/XOFF flow control has been enabled or disabled
This method is called to notify the client whether or not
to enable XON/XOFF flow control. If client_can_do is
`True` and output is being sent to an interactive
terminal the application should allow input of Control-S
and Control-Q to pause and resume output, respectively.
If client_can_do is `False`, Control-S and Control-Q
should be treated as normal input and passed through to
the server. Non-interactive applications can ignore this
request.
By default, this message is ignored.
:param client_can_do:
Whether or not to enable XON/XOFF flow control
:type client_can_do: `bool`
"""
[docs]
def exit_status_received(self, status: int) -> None:
"""A remote exit status has been received for this session
This method is called when the shell, command, or subsystem
running on the server terminates and returns an exit status.
A zero exit status generally means that the operation was
successful. This call will generally be followed by a call
to :meth:`connection_lost`.
By default, the exit status is ignored.
:param status:
The exit status returned by the remote process
:type status: `int`
"""
[docs]
def exit_signal_received(self, signal: str, core_dumped: bool,
msg: str, lang: str) -> None:
"""A remote exit signal has been received for this session
This method is called when the shell, command, or subsystem
running on the server terminates abnormally with a signal.
A more detailed error may also be provided, along with an
indication of whether the remote process dumped core. This call
will generally be followed by a call to :meth:`connection_lost`.
By default, exit signals are ignored.
:param signal:
The signal which caused the remote process to exit
:param core_dumped:
Whether or not the remote process dumped core
:param msg:
Details about what error occurred
:param lang:
The language the error message is in
:type signal: `str`
:type core_dumped: `bool`
:type msg: `str`
:type lang: `str`
"""
[docs]
class SSHServerSession(SSHSession[AnyStr]):
"""SSH server session handler
Applications should subclass this when implementing an SSH server
session handler. The functions listed below should be implemented
to define application-specific behavior. In particular, the
standard `asyncio` protocol methods such as :meth:`connection_made`,
:meth:`connection_lost`, :meth:`data_received`, :meth:`eof_received`,
:meth:`pause_writing`, and :meth:`resume_writing` are all supported.
In addition, :meth:`pty_requested` is called when the client requests a
pseudo-terminal, one of :meth:`shell_requested`, :meth:`exec_requested`,
or :meth:`subsystem_requested` is called depending on what type of
session the client wants to start, :meth:`session_started` is called
once the SSH session is fully started, :meth:`terminal_size_changed` is
called when the client's terminal size changes, :meth:`signal_received`
is called when the client sends a signal, and :meth:`break_received`
is called when the client sends a break.
"""
# pylint: disable=no-self-use,unused-argument
[docs]
def connection_made(self, chan: 'SSHServerChannel[AnyStr]') -> None:
"""Called when a channel is opened successfully
This method is called when a channel is opened successfully. The
channel parameter should be stored if needed for later use.
:param chan:
The channel which was successfully opened.
:type chan: :class:`SSHServerChannel`
"""
[docs]
def pty_requested(self, term_type: str,
term_size: Tuple[int, int, int, int],
term_modes: Mapping[int, int]) -> bool:
"""A pseudo-terminal has been requested
This method is called when the client sends a request to allocate
a pseudo-terminal with the requested terminal type, size, and
POSIX terminal modes. This method should return `True` if the
request for the pseudo-terminal is accepted. Otherwise, it should
return `False` to reject the request.
By default, requests to allocate a pseudo-terminal are accepted
but nothing is done with the associated terminal information.
Applications wishing to use this information should implement
this method and have it return `True`, or call
:meth:`get_terminal_type() <SSHServerChannel.get_terminal_type>`,
:meth:`get_terminal_size() <SSHServerChannel.get_terminal_size>`,
or :meth:`get_terminal_mode() <SSHServerChannel.get_terminal_mode>`
on the :class:`SSHServerChannel` to get the information they need
after a shell, command, or subsystem is started.
:param term_type:
Terminal type to set for this session
:param term_size:
Terminal size to set for this session provided as a
tuple of four `int` values: the width and height of the
terminal in characters followed by the width and height
of the terminal in pixels
:param term_modes:
POSIX terminal modes to set for this session, where keys
are taken from :ref:`POSIX terminal modes <PTYModes>` with
values defined in section 8 of :rfc:`RFC 4254 <4254#section-8>`.
:type term_type: `str`
:type term_size: tuple of 4 `int` values
:type term_modes: `dict`
:returns: A `bool` indicating if the request for a
pseudo-terminal was allowed or not
"""
return True # pragma: no cover
[docs]
def terminal_size_changed(self, width: int, height: int,
pixwidth: int, pixheight: int) -> None:
"""The terminal size has changed
This method is called when a client requests a
pseudo-terminal and again whenever the the size of
he client's terminal window changes.
By default, this information is ignored, but applications
wishing to use the terminal size can implement this method
to get notified whenever it changes.
:param width:
The width of the terminal in characters
:param height:
The height of the terminal in characters
:param pixwidth: (optional)
The width of the terminal in pixels
:param pixheight: (optional)
The height of the terminal in pixels
:type width: `int`
:type height: `int`
:type pixwidth: `int`
:type pixheight: `int`
"""
[docs]
def shell_requested(self) -> bool:
"""The client has requested a shell
This method should be implemented by the application to
perform whatever processing is required when a client makes
a request to open an interactive shell. It should return
`True` to accept the request, or `False` to reject it.
If the application returns `True`, the :meth:`session_started`
method will be called once the channel is fully open. No output
should be sent until this method is called.
By default this method returns `False` to reject all requests.
:returns: A `bool` indicating if the shell request was
allowed or not
"""
return False # pragma: no cover
[docs]
def exec_requested(self, command: str) -> bool:
"""The client has requested to execute a command
This method should be implemented by the application to
perform whatever processing is required when a client makes
a request to execute a command. It should return `True` to
accept the request, or `False` to reject it.
If the application returns `True`, the :meth:`session_started`
method will be called once the channel is fully open. No output
should be sent until this method is called.
By default this method returns `False` to reject all requests.
:param command:
The command the client has requested to execute
:type command: `str`
:returns: A `bool` indicating if the exec request was
allowed or not
"""
return False # pragma: no cover
[docs]
def subsystem_requested(self, subsystem: str) -> bool:
"""The client has requested to start a subsystem
This method should be implemented by the application to
perform whatever processing is required when a client makes
a request to start a subsystem. It should return `True` to
accept the request, or `False` to reject it.
If the application returns `True`, the :meth:`session_started`
method will be called once the channel is fully open. No output
should be sent until this method is called.
By default this method returns `False` to reject all requests.
:param subsystem:
The subsystem to start
:type subsystem: `str`
:returns: A `bool` indicating if the request to open the
subsystem was allowed or not
"""
return False # pragma: no cover
[docs]
def break_received(self, msec: int) -> bool:
"""The client has sent a break
This method is called when the client requests that the
server perform a break operation on the terminal. If the
break is performed, this method should return `True`.
Otherwise, it should return `False`.
By default, this method returns `False` indicating that
no break was performed.
:param msec:
The duration of the break in milliseconds
:type msec: `int`
:returns: A `bool` to indicate if the break operation was
performed or not
"""
return False # pragma: no cover
[docs]
def signal_received(self, signal: str) -> None:
"""The client has sent a signal
This method is called when the client delivers a signal
on the channel.
By default, signals from the client are ignored.
:param signal:
The name of the signal received
:type signal: `str`
"""
def soft_eof_received(self) -> None:
"""The client has sent a soft EOF
This method is called by the line editor when the client
send a soft EOF (Ctrl-D on an empty input line).
By default, soft EOF will trigger an EOF to an outstanding
read call but still allow additional input to be received
from the client after that.
"""
[docs]
class SSHTCPSession(SSHSession[AnyStr]):
"""SSH TCP session handler
Applications should subclass this when implementing a handler for
SSH direct or forwarded TCP connections.
SSH client applications wishing to open a direct connection should call
:meth:`create_connection() <SSHClientConnection.create_connection>`
on their :class:`SSHClientConnection`, passing in a factory which
returns instances of this class.
Server applications wishing to allow direct connections should
implement the coroutine :meth:`connection_requested()
<SSHServer.connection_requested>` on their :class:`SSHServer`
object and have it return instances of this class.
Server applications wishing to allow connection forwarding back
to the client should implement the coroutine :meth:`server_requested()
<SSHServer.server_requested>` on their :class:`SSHServer` object
and call :meth:`create_connection()
<SSHServerConnection.create_connection>` on their
:class:`SSHServerConnection` for each new connection, passing it a
factory which returns instances of this class.
When a connection is successfully opened, :meth:`session_started`
will be called, after which the application can begin sending data.
Received data will be passed to the :meth:`data_received` method.
"""
[docs]
def connection_made(self, chan: 'SSHTCPChannel[AnyStr]') -> None:
"""Called when a channel is opened successfully
This method is called when a channel is opened successfully. The
channel parameter should be stored if needed for later use.
:param chan:
The channel which was successfully opened.
:type chan: :class:`SSHTCPChannel`
"""
[docs]
class SSHUNIXSession(SSHSession[AnyStr]):
"""SSH UNIX domain socket session handler
Applications should subclass this when implementing a handler for
SSH direct or forwarded UNIX domain socket connections.
SSH client applications wishing to open a direct connection should call
:meth:`create_unix_connection()
<SSHClientConnection.create_unix_connection>` on their
:class:`SSHClientConnection`, passing in a factory which returns
instances of this class.
Server applications wishing to allow direct connections should
implement the coroutine :meth:`unix_connection_requested()
<SSHServer.unix_connection_requested>` on their :class:`SSHServer`
object and have it return instances of this class.
Server applications wishing to allow connection forwarding back
to the client should implement the coroutine
:meth:`unix_server_requested() <SSHServer.unix_server_requested>`
on their :class:`SSHServer` object and call
:meth:`create_unix_connection()
<SSHServerConnection.create_unix_connection>` on their
:class:`SSHServerConnection` for each new connection, passing it a
factory which returns instances of this class.
When a connection is successfully opened, :meth:`session_started`
will be called, after which the application can begin sending data.
Received data will be passed to the :meth:`data_received` method.
"""
[docs]
def connection_made(self, chan: 'SSHUNIXChannel[AnyStr]') -> None:
"""Called when a channel is opened successfully
This method is called when a channel is opened successfully. The
channel parameter should be stored if needed for later use.
:param chan:
The channel which was successfully opened.
:type chan: :class:`SSHUNIXChannel`
"""
[docs]
class SSHTunTapSession(SSHSession[bytes]):
"""SSH TUN/TAP session handler
Applications should subclass this when implementing a handler for
SSH TUN/TAP tunnels.
SSH client applications wishing to open a tunnel should call
:meth:`create_tun() <SSHClientConnection.create_tun>` or
:meth:`create_tap() <SSHClientConnection.create_tap>` on their
:class:`SSHClientConnection`, passing in a factory which returns
instances of this class.
Server applications wishing to allow tunnel connections should
implement the coroutine :meth:`tun_requested()
<SSHServer.tun_requested>` or :meth:`tap_requested()
<SSHServer.tap_requested>` on their :class:`SSHServer` object
and have it return instances of this class.
When a connection is successfully opened, :meth:`session_started`
will be called, after which the application can begin sending data.
Received data will be passed to the :meth:`data_received` method.
"""
[docs]
def connection_made(self, chan: 'SSHTunTapChannel') -> None:
"""Called when a channel is opened successfully
This method is called when a channel is opened successfully. The
channel parameter should be stored if needed for later use.
:param chan:
The channel which was successfully opened.
:type chan: :class:`SSHTunTapChannel`
"""
SSHSessionFactory = Callable[[], SSHSession[AnyStr]]
SSHClientSessionFactory = Callable[[], SSHClientSession[AnyStr]]
SSHTCPSessionFactory = Callable[[], SSHTCPSession[AnyStr]]
SSHUNIXSessionFactory = Callable[[], SSHUNIXSession[AnyStr]]
SSHTunTapSessionFactory = Callable[[], SSHTunTapSession]