Module scrapli.transport.plugins.telnet.transport
scrapli.transport.plugins.telnet.transport
Expand source code
"""scrapli.transport.plugins.telnet.transport"""
from dataclasses import dataclass
from telnetlib import Telnet
from typing import Optional
from scrapli.decorators import TransportTimeout
from scrapli.exceptions import ScrapliConnectionError, ScrapliConnectionNotOpened
from scrapli.transport.base import BasePluginTransportArgs, BaseTransportArgs, Transport
@dataclass()
class PluginTransportArgs(BasePluginTransportArgs):
pass
class ScrapliTelnet(Telnet):
def __init__(self, host: str, port: int, timeout: float) -> None:
"""
ScrapliTelnet class for typing purposes
Args:
host: string of host
port: integer port to connect to
timeout: timeout value in seconds
Returns:
None
Raises:
N/A
"""
self.eof: bool
self.timeout: float
super().__init__(host, port, int(timeout))
class TelnetTransport(Transport):
def __init__(
self, base_transport_args: BaseTransportArgs, plugin_transport_args: PluginTransportArgs
) -> None:
super().__init__(base_transport_args=base_transport_args)
self.plugin_transport_args = plugin_transport_args
self.session: Optional[ScrapliTelnet] = None
def open(self) -> None:
self._pre_open_closing_log(closing=False)
# establish session with "socket" timeout, then reset timeout to "transport" timeout
try:
self.session = ScrapliTelnet(
host=self._base_transport_args.host,
port=self._base_transport_args.port,
timeout=self._base_transport_args.timeout_socket,
)
self.session.timeout = self._base_transport_args.timeout_transport
except ConnectionError as exc:
msg = f"Failed to open telnet session to host {self._base_transport_args.host}"
if "connection refused" in str(exc).lower():
msg = (
f"Failed to open telnet session to host {self._base_transport_args.host}, "
"connection refused"
)
raise ScrapliConnectionError(msg) from exc
self._post_open_closing_log(closing=False)
def close(self) -> None:
self._pre_open_closing_log(closing=True)
if self.session:
self.session.close()
self.session = None
self._post_open_closing_log(closing=True)
def isalive(self) -> bool:
if not self.session:
return False
return not self.session.eof
@TransportTimeout("timed out reading from transport")
def read(self) -> bytes:
if not self.session:
raise ScrapliConnectionNotOpened
try:
buf = self.session.read_eager()
except Exception as exc:
raise ScrapliConnectionError(
"encountered EOF reading from transport; typically means the device closed the "
"connection"
) from exc
return buf
def write(self, channel_input: bytes) -> None:
if not self.session:
raise ScrapliConnectionNotOpened
self.session.write(channel_input)
Classes
PluginTransportArgs
Expand source code
class PluginTransportArgs(BasePluginTransportArgs):
pass
Ancestors (in MRO)
- scrapli.transport.base.base_transport.BasePluginTransportArgs
ScrapliTelnet
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63 | Telnet interface class.
An instance of this class represents a connection to a telnet
server. The instance is initially not connected; the open()
method must be used to establish a connection. Alternatively, the
host name and optional port number can be passed to the
constructor, too.
Don't try to reopen an already connected instance.
This class has many read_*() methods. Note that some of them
raise EOFError when the end of the connection is read, because
they can return an empty string for other reasons. See the
individual doc strings.
read_until(expected, [timeout])
Read until the expected string has been seen, or a timeout is
hit (default is no timeout); may block.
read_all()
Read all data until EOF; may block.
read_some()
Read at least one byte or EOF; may block.
read_very_eager()
Read all data available already queued or on the socket,
without blocking.
read_eager()
Read either data already queued or some data available on the
socket, without blocking.
read_lazy()
Read all data in the raw queue (processing it first), without
doing any socket I/O.
read_very_lazy()
Reads all data in the cooked queue, without doing any socket
I/O.
read_sb_data()
Reads available data between SB ... SE sequence. Don't block.
set_option_negotiation_callback(callback)
Each time a telnet option is read on the input flow, this callback
(if set) is called with the following parameters :
callback(telnet socket, command, option)
option will be chr(0) when there is no option.
No other action is done afterwards by telnetlib.
ScrapliTelnet class for typing purposes
Args:
host: string of host
port: integer port to connect to
timeout: timeout value in seconds
Returns:
None
Raises:
N/A
|
Expand source code
class ScrapliTelnet(Telnet):
def __init__(self, host: str, port: int, timeout: float) -> None:
"""
ScrapliTelnet class for typing purposes
Args:
host: string of host
port: integer port to connect to
timeout: timeout value in seconds
Returns:
None
Raises:
N/A
"""
self.eof: bool
self.timeout: float
super().__init__(host, port, int(timeout))
Ancestors (in MRO)
TelnetTransport
| Helper class that provides a standard way to create an ABC using
inheritance.
|
Expand source code
class TelnetTransport(Transport):
def __init__(
self, base_transport_args: BaseTransportArgs, plugin_transport_args: PluginTransportArgs
) -> None:
super().__init__(base_transport_args=base_transport_args)
self.plugin_transport_args = plugin_transport_args
self.session: Optional[ScrapliTelnet] = None
def open(self) -> None:
self._pre_open_closing_log(closing=False)
# establish session with "socket" timeout, then reset timeout to "transport" timeout
try:
self.session = ScrapliTelnet(
host=self._base_transport_args.host,
port=self._base_transport_args.port,
timeout=self._base_transport_args.timeout_socket,
)
self.session.timeout = self._base_transport_args.timeout_transport
except ConnectionError as exc:
msg = f"Failed to open telnet session to host {self._base_transport_args.host}"
if "connection refused" in str(exc).lower():
msg = (
f"Failed to open telnet session to host {self._base_transport_args.host}, "
"connection refused"
)
raise ScrapliConnectionError(msg) from exc
self._post_open_closing_log(closing=False)
def close(self) -> None:
self._pre_open_closing_log(closing=True)
if self.session:
self.session.close()
self.session = None
self._post_open_closing_log(closing=True)
def isalive(self) -> bool:
if not self.session:
return False
return not self.session.eof
@TransportTimeout("timed out reading from transport")
def read(self) -> bytes:
if not self.session:
raise ScrapliConnectionNotOpened
try:
buf = self.session.read_eager()
except Exception as exc:
raise ScrapliConnectionError(
"encountered EOF reading from transport; typically means the device closed the "
"connection"
) from exc
return buf
def write(self, channel_input: bytes) -> None:
if not self.session:
raise ScrapliConnectionNotOpened
self.session.write(channel_input)
Ancestors (in MRO)
- scrapli.transport.base.sync_transport.Transport
- scrapli.transport.base.base_transport.BaseTransport
- abc.ABC