Skip to content

transport

scrapli.transport.plugins.system.transport

SystemTransport

Bases: Transport

Source code in transport/plugins/system/transport.py 
 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
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class SystemTransport(Transport):
    def __init__(
        self, base_transport_args: BaseTransportArgs, plugin_transport_args: PluginTransportArgs
    ) -> None:
        """
        System (i.e. /bin/ssh) transport plugin.

        This transport supports some additional `transport_options` to control behavior --

        `ptyprocess` is a dictionary that has the following options:
            rows: integer number of rows for ptyprocess "window"
            cols: integer number of cols for ptyprocess "window"
            echo: defaults to `True`, passing `False` disables echo in the ptyprocess; should only
                be used with scrapli-netconf, will break scrapli!

        `netconf_force_pty` is a scrapli-netconf only argument. This setting defaults to `True` and
            allows you to *not* force a pty. This setting seems to only be necessary when connecting
            to juniper devices on port 830 as junos decides to not allocate a pty on that port for
            some reason!

        Args:
            base_transport_args: scrapli base transport plugin arguments
            plugin_transport_args: system ssh specific transport plugin arguments

        Returns:
            N/A

        Raises:
            ScrapliUnsupportedPlatform: if system is windows

        """
        super().__init__(base_transport_args=base_transport_args)
        self.plugin_transport_args = plugin_transport_args

        if sys.platform.startswith("win"):
            raise ScrapliUnsupportedPlatform("system transport is not supported on windows devices")

        self.open_cmd: List[str] = []
        self.session: Optional[PtyProcess] = None

    def _build_open_cmd(self) -> None:
        """
        Method to craft command to open ssh session

        Args:
            N/A

        Returns:
            None

        Raises:
            N/A

        """
        if self.open_cmd:
            self.open_cmd = []

        self.open_cmd.extend(["ssh", self._base_transport_args.host])
        self.open_cmd.extend(["-p", str(self._base_transport_args.port)])

        self.open_cmd.extend(
            ["-o", f"ConnectTimeout={int(self._base_transport_args.timeout_socket)}"]
        )
        self.open_cmd.extend(
            ["-o", f"ServerAliveInterval={int(self._base_transport_args.timeout_transport)}"]
        )

        if self.plugin_transport_args.auth_private_key:
            self.open_cmd.extend(["-i", self.plugin_transport_args.auth_private_key])
        if self.plugin_transport_args.auth_username:
            self.open_cmd.extend(["-l", self.plugin_transport_args.auth_username])

        if self.plugin_transport_args.auth_strict_key is False:
            self.open_cmd.extend(["-o", "StrictHostKeyChecking=no"])
            self.open_cmd.extend(["-o", "UserKnownHostsFile=/dev/null"])
        else:
            self.open_cmd.extend(["-o", "StrictHostKeyChecking=yes"])
            if self.plugin_transport_args.ssh_known_hosts_file:
                self.open_cmd.extend(
                    ["-o", f"UserKnownHostsFile={self.plugin_transport_args.ssh_known_hosts_file}"]
                )

        if self.plugin_transport_args.ssh_config_file:
            self.open_cmd.extend(["-F", self.plugin_transport_args.ssh_config_file])
        else:
            self.open_cmd.extend(["-F", "/dev/null"])

        open_cmd_user_args = self._base_transport_args.transport_options.get("open_cmd", [])
        if isinstance(open_cmd_user_args, str):
            open_cmd_user_args = [open_cmd_user_args]
        self.open_cmd.extend(open_cmd_user_args)

        self.logger.debug(f"created transport 'open_cmd': '{self.open_cmd}'")

    def open(self) -> None:
        self._pre_open_closing_log(closing=False)

        if not self.open_cmd:
            self._build_open_cmd()

        self.session = PtyProcess.spawn(
            self.open_cmd,
            echo=self._base_transport_args.transport_options.get("ptyprocess", {}).get(
                "echo", True
            ),
            rows=self._base_transport_args.transport_options.get("ptyprocess", {}).get("rows", 80),
            cols=self._base_transport_args.transport_options.get("ptyprocess", {}).get("cols", 256),
        )

        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
        if self.session.isalive() and not self.session.eof():
            return True
        return False

    @timeout_wrapper
    def read(self) -> bytes:
        if not self.session:
            raise ScrapliConnectionNotOpened
        try:
            buf = self.session.read(65535)
        except EOFError as exc:
            msg = (
                "encountered EOF reading from transport; typically means the device closed the "
                "connection"
            )
            self.logger.critical(msg)
            raise ScrapliConnectionError(msg) from exc

        return buf

    def write(self, channel_input: bytes) -> None:
        if not self.session:
            raise ScrapliConnectionNotOpened
        self.session.write(channel_input)

__init__(base_transport_args: BaseTransportArgs, plugin_transport_args: PluginTransportArgs) -> None

System (i.e. /bin/ssh) transport plugin.

This transport supports some additional transport_options to control behavior --

ptyprocess is a dictionary that has the following options: rows: integer number of rows for ptyprocess "window" cols: integer number of cols for ptyprocess "window" echo: defaults to True, passing False disables echo in the ptyprocess; should only be used with scrapli-netconf, will break scrapli!

netconf_force_pty is a scrapli-netconf only argument. This setting defaults to True and allows you to not force a pty. This setting seems to only be necessary when connecting to juniper devices on port 830 as junos decides to not allocate a pty on that port for some reason!

Parameters:

Name Type Description Default
base_transport_args BaseTransportArgs

scrapli base transport plugin arguments

 required
plugin_transport_args PluginTransportArgs

system ssh specific transport plugin arguments

 required

Returns:

Type Description
None

N/A

Raises:

Type Description
ScrapliUnsupportedPlatform

if system is windows
Source code in transport/plugins/system/transport.py 
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
def __init__(
    self, base_transport_args: BaseTransportArgs, plugin_transport_args: PluginTransportArgs
) -> None:
    """
    System (i.e. /bin/ssh) transport plugin.

    This transport supports some additional `transport_options` to control behavior --

    `ptyprocess` is a dictionary that has the following options:
        rows: integer number of rows for ptyprocess "window"
        cols: integer number of cols for ptyprocess "window"
        echo: defaults to `True`, passing `False` disables echo in the ptyprocess; should only
            be used with scrapli-netconf, will break scrapli!

    `netconf_force_pty` is a scrapli-netconf only argument. This setting defaults to `True` and
        allows you to *not* force a pty. This setting seems to only be necessary when connecting
        to juniper devices on port 830 as junos decides to not allocate a pty on that port for
        some reason!

    Args:
        base_transport_args: scrapli base transport plugin arguments
        plugin_transport_args: system ssh specific transport plugin arguments

    Returns:
        N/A

    Raises:
        ScrapliUnsupportedPlatform: if system is windows

    """
    super().__init__(base_transport_args=base_transport_args)
    self.plugin_transport_args = plugin_transport_args

    if sys.platform.startswith("win"):
        raise ScrapliUnsupportedPlatform("system transport is not supported on windows devices")

    self.open_cmd: List[str] = []
    self.session: Optional[PtyProcess] = None