Built-in Plugins API

NOS Plugins

NOS plugins are at the heart of SIMNOS, they are what enables to realize its full potential.

Cisco IOS

NOS module for Cisco IOS

CiscoIOS

Bases: BaseDevice

Class that keeps track of the state of the Cisco IOS device.

Source code in simnos/plugins/nos/platforms_py/cisco_ios.py

class CiscoIOS(BaseDevice):
    """
    Class that keeps track of the state of the Cisco IOS device.
    """

    def make_show_clock(self, base_prompt, current_prompt, command):
        "Return String in format '*11:54:03.018 UTC Sat Apr 16 2022'"
        return time.strftime("*%H:%M:%S.000 %Z %a %b %d %Y")

    def make_show_running_config(self, base_prompt, current_prompt, command):
        "Return String of running configuration"
        return self.render("cisco_ios/show_running-config.j2", base_prompt=base_prompt)

    def make_show_version(self, base_prompt, current_prompt, command):
        "Return String of system hardware and software status"
        return self.render("cisco_ios/show_version.j2", base_prompt=base_prompt)

make_show_clock(base_prompt, current_prompt, command)

Return String in format '*11:54:03.018 UTC Sat Apr 16 2022'

Source code in simnos/plugins/nos/platforms_py/cisco_ios.py

def make_show_clock(self, base_prompt, current_prompt, command):
    "Return String in format '*11:54:03.018 UTC Sat Apr 16 2022'"
    return time.strftime("*%H:%M:%S.000 %Z %a %b %d %Y")

make_show_running_config(base_prompt, current_prompt, command)

Return String of running configuration

Source code in simnos/plugins/nos/platforms_py/cisco_ios.py

def make_show_running_config(self, base_prompt, current_prompt, command):
    "Return String of running configuration"
    return self.render("cisco_ios/show_running-config.j2", base_prompt=base_prompt)

make_show_version(base_prompt, current_prompt, command)

Return String of system hardware and software status

Source code in simnos/plugins/nos/platforms_py/cisco_ios.py

def make_show_version(self, base_prompt, current_prompt, command):
    "Return String of system hardware and software status"
    return self.render("cisco_ios/show_version.j2", base_prompt=base_prompt)

Server Plugins

Server plugins act as an access layer, simulating device connections.

ParamikoSshServer

Bases: TCPServerBase

Class to implement an SSH server using paramiko as the SSH connection library.

Source code in simnos/plugins/servers/ssh_server_paramiko.py

class ParamikoSshServer(TCPServerBase):
    """
    Class to implement an SSH server using paramiko
    as the SSH connection library.
    """

    _moduli_loaded: bool | None = None
    _moduli_lock: threading.Lock = threading.Lock()
    _default_key: paramiko.rsakey.RSAKey | None = None
    _default_key_lock: threading.Lock = threading.Lock()
    _KNOWN_KEY_TYPES = (
        "ssh-rsa",
        "ssh-ed25519",
        "ssh-dss",
        "ecdsa-sha2-",
        "sk-ssh-ed25519",
        "sk-ecdsa-sha2-",
    )

    def __init__(
        self,
        shell: type,
        nos: Nos,
        nos_inventory_config: dict,
        port: int,
        username: str,
        password: str,
        ssh_key_file: str | None = None,
        ssh_key_file_password: str | None = None,
        ssh_banner: str = "SIMNOS Paramiko SSH Server",
        shell_configuration: dict | None = None,
        address: str = "127.0.0.1",
        timeout: int = 1,
        watchdog_interval: float = 1,
        authorized_keys: str | None = None,
    ):
        super().__init__(address=address, port=port, timeout=timeout)

        self.nos: Nos = nos
        self.nos_inventory_config: dict = nos_inventory_config
        self.shell: type = shell
        self.shell_configuration: dict = shell_configuration or {}
        self.ssh_banner: str = ssh_banner
        self.username: str = username
        self.password: str = password
        self.watchdog_interval: float = watchdog_interval
        self._authorized_keys = self._load_authorized_keys(authorized_keys) if authorized_keys else None

        if ssh_key_file:
            self._ssh_server_key: paramiko.rsakey.RSAKey = paramiko.RSAKey.from_private_key_file(
                ssh_key_file, ssh_key_file_password
            )
        else:
            with ParamikoSshServer._default_key_lock:
                if ParamikoSshServer._default_key is None:
                    ParamikoSshServer._default_key = paramiko.RSAKey.generate(2048)
            self._ssh_server_key = ParamikoSshServer._default_key
            log.warning(
                "Using auto-generated SSH host key. This key is not persisted and "
                "will change on restart. Provide a custom key via ssh_key_file "
                "for non-local use."
            )

        # Load SSH moduli once for DH Group Exchange (GEX) support in server mode.
        # Prefer system moduli (live, distro-rotated) when available; fall back to
        # the moduli file bundled with the package on hosts without /etc/ssh/moduli
        # (Windows / macOS). Result is cached at the class level under a lock.
        #
        # `_moduli_lock` only serializes SIMNOS-internal init; the underlying
        # `paramiko.Transport._modulus_pack` is a paramiko-global state and is
        # not lockable from here. If another thread loads moduli via paramiko
        # directly (outside SIMNOS), it can still race. Mirrors the
        # `_default_key_lock` pattern in scope, not in coverage.
        with ParamikoSshServer._moduli_lock:
            if ParamikoSshServer._moduli_loaded is None:
                ok = paramiko.Transport.load_server_moduli()
                if not ok:
                    # `is_file()` returns False for missing path, directory, or
                    # broken symlink. The latter two are extreme edge cases for
                    # a package-bundled file; treating them as "missing" is fine.
                    if _BUNDLED_MODULI.is_file():
                        ok = paramiko.Transport.load_server_moduli(filename=str(_BUNDLED_MODULI))
                        if not ok:
                            log.error(
                                "Bundled moduli at %s exists but failed to load "
                                "(possibly corrupted or unreadable). Falling back "
                                "to GEX-disable workaround.",
                                _BUNDLED_MODULI,
                            )
                    else:
                        log.error(
                            "Bundled moduli file missing at %s — falling back to "
                            "GEX-disable workaround. This indicates a packaging "
                            "regression, please report.",
                            _BUNDLED_MODULI,
                        )
                ParamikoSshServer._moduli_loaded = ok

    @staticmethod
    def _load_authorized_keys(path: str) -> set[tuple[str, str]]:
        """Parse an OpenSSH authorized_keys file.

        Supports bare key lines and lines with leading options.
        Skips comment lines, blank lines, and @marker lines.
        File not found / permission errors propagate as-is (fail-fast).

        Returns a set of (key_type, base64_data) tuples.
        """
        keys: set[tuple[str, str]] = set()
        with open(path) as fh:
            for line in fh:
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                if line.startswith("@"):
                    log.warning("Skipping unsupported marker line: %s", line)
                    continue
                parts = line.split()
                for i, part in enumerate(parts):
                    if any(part.startswith(prefix) for prefix in ParamikoSshServer._KNOWN_KEY_TYPES):
                        if i + 1 < len(parts):
                            keys.add((part, parts[i + 1]))
                        else:
                            log.warning("Key type found but base64 data missing, skipping line: %s", line)
                        break
                else:
                    log.warning("No known key type found, skipping line: %s", line)
        return keys

    def watchdog(
        self,
        is_running: threading.Event,
        run_srv: threading.Event,
        session: paramiko.Transport,
        shell: Any,
    ):
        """
        Method to monitor server liveness and recover where possible.
        """
        while run_srv.is_set():
            if not session.is_alive():
                log.warning("ParamikoSshServer.watchdog - session not alive, stopping shell")
                break

            if not is_running.is_set():
                break

            time.sleep(min(self.watchdog_interval, SHUTDOWN_IO_TIMEOUT))

        shell.stop()

    def _channel_login(self, channel) -> tuple[bool, bool]:
        """
        Perform channel-level login for auth_none platforms (e.g. Dell PowerConnect).

        Thin wrapper around tap_bridge.interactive_login supplying the
        Dell-style prompts; the interaction itself is shared with Telnet.
        Expects the channel timeout to be configured by the caller
        (connection_function) beforehand.

        :param channel: paramiko Channel
        :return: (authenticated, skip_lf) — skip_lf should be forwarded to
                 client_to_shell_tap so it can consume a trailing LF after
                 the final CR of the password line.
        """
        return interactive_login(
            ParamikoChannelAdapter(channel),
            self.username,
            self.password,
            user_prompt=b"\r\nUser Name:",
            pass_prompt=b"\r\nPassword:",
        )

    def connection_function(self, client: socket.socket, is_running: threading.Event):
        shell_replied_event = threading.Event()
        run_srv = threading.Event()
        run_srv.set()

        # determine if this NOS requires auth_none
        allow_auth_none = getattr(self.nos, "auth", None) == "none"

        # create the SSH transport object
        session = paramiko.Transport(client)
        if not ParamikoSshServer._moduli_loaded:
            session.disabled_algorithms = _DISABLED_GEX_ALGORITHMS
        session.add_server_key(self._ssh_server_key)
        session.banner_timeout = SHUTDOWN_IO_TIMEOUT
        session.handshake_timeout = SHUTDOWN_IO_TIMEOUT

        try:
            # create the server
            server = ParamikoSshServerInterface(
                ssh_banner=self.ssh_banner,
                username=self.username,
                password=self.password,
                allow_auth_none=allow_auth_none,
                authorized_keys=self._authorized_keys,
            )

            # start the SSH server — may raise SSHException if the client
            # disconnects during handshake or if stop() races with accept.
            try:
                session.start_server(server=server)
            except paramiko.SSHException as e:
                log.debug("SSH handshake failed (likely client disconnect or stop): %s", e)
                return

            # wait for the client to open a channel
            channel = None
            while channel is None and is_running.is_set() and session.is_alive():
                channel = session.accept(SHUTDOWN_IO_TIMEOUT)
            if channel is None:
                log.warning("session.accept() returned None or server stopping, closing transport")
                return

            # Timeout responsibility lives here (not in the adapter / shared
            # helpers): configure it before any channel I/O below.
            channel.settimeout(self.timeout)

            # For auth_none platforms (e.g. Dell PowerConnect), perform channel-level
            # login before starting the shell.  When publickey auth is also configured,
            # clients that authenticate via publickey bypass this channel-level login
            # intentionally — SSH-level publickey auth already verified the identity.
            skip_lf = False
            if server.auth_method_used == "none":
                try:
                    authenticated, skip_lf = self._channel_login(channel)
                except (TimeoutError, OSError, EOFError, paramiko.SSHException):
                    log.debug("Client disconnected during channel login")
                    return
                if not authenticated:
                    log.warning("Channel login failed, closing connection")
                    return

            # create stdio for the shell
            shell_stdin, shell_stdout = TapIO(run_srv), TapIO(run_srv)

            # bridge the channel and the shell through the shared tap pair
            transport_adapter = ParamikoChannelAdapter(channel)

            # start intermediate thread to tap into
            # the client->shell_stdin bytes stream
            client_to_shell_tapper = threading.Thread(
                target=client_to_shell_tap,
                args=(transport_adapter, shell_stdin, shell_replied_event, run_srv),
                kwargs={"initial_skip_lf": skip_lf, "shell_stdout": shell_stdout},
                daemon=True,
            )
            client_to_shell_tapper.start()

            # start intermediate thread to tap into
            # the shell_stdout->client bytes stream
            shell_to_client_tapper = threading.Thread(
                target=shell_to_client_tap,
                args=(transport_adapter, shell_stdout, shell_replied_event, run_srv),
                daemon=True,
            )
            shell_to_client_tapper.start()

            # create the client shell
            client_shell = self.shell(
                stdin=shell_stdin,
                stdout=shell_stdout,
                nos=self.nos,
                nos_inventory_config=self.nos_inventory_config,
                is_running=is_running,
                **self.shell_configuration,
            )

            # start watchdog thread
            watchdog_thread = threading.Thread(
                target=self.watchdog, args=(is_running, run_srv, session, client_shell), daemon=True
            )
            watchdog_thread.start()

            # running this command will block this function until shell exits
            client_shell.start()
            log.debug("ParamikoSshServer.connection_function stopped shell thread")

        finally:
            # Stop all server threads
            run_srv.clear()
            log.debug("ParamikoSshServer.connection_function stopped server threads")

            session.close()
            log.debug("ParamikoSshServer.connection_function closed transport %s", session)

watchdog(is_running, run_srv, session, shell)

Method to monitor server liveness and recover where possible.

Source code in simnos/plugins/servers/ssh_server_paramiko.py

def watchdog(
    self,
    is_running: threading.Event,
    run_srv: threading.Event,
    session: paramiko.Transport,
    shell: Any,
):
    """
    Method to monitor server liveness and recover where possible.
    """
    while run_srv.is_set():
        if not session.is_alive():
            log.warning("ParamikoSshServer.watchdog - session not alive, stopping shell")
            break

        if not is_running.is_set():
            break

        time.sleep(min(self.watchdog_interval, SHUTDOWN_IO_TIMEOUT))

    shell.stop()

TelnetServer

Bases: TCPServerBase

Telnet server plugin using raw sockets.

Follows the same plugin architecture as ParamikoSshServer: TCPServerBase → connection_function() → TapIO → CMDShell.

Source code in simnos/plugins/servers/telnet_server.py

class TelnetServer(TCPServerBase):
    """
    Telnet server plugin using raw sockets.

    Follows the same plugin architecture as ParamikoSshServer:
    TCPServerBase → connection_function() → TapIO → CMDShell.
    """

    def __init__(
        self,
        shell: type,
        nos: Nos,
        nos_inventory_config: dict,
        port: int,
        username: str,
        password: str,
        banner: str = "SIMNOS Telnet Server",
        shell_configuration: dict | None = None,
        address: str = "127.0.0.1",
        timeout: int = 1,
        watchdog_interval: float = 1,
    ):
        super().__init__(address=address, port=port, timeout=timeout)

        self.nos: Nos = nos
        self.nos_inventory_config: dict = nos_inventory_config
        self.shell: type = shell
        self.shell_configuration: dict = shell_configuration or {}
        self.banner: str = banner
        self.username: str = username
        self.password: str = password
        self.watchdog_interval: float = watchdog_interval

        if not _is_loopback(address):
            log.warning(
                "Telnet transmits all data (including credentials) in plaintext. "
                "Binding to non-local address %s is insecure. "
                "Use SSH (ParamikoSshServer) for non-local access.",
                address,
            )

    # ------------------------------------------------------------------
    # IAC handling
    # ------------------------------------------------------------------

    def _recv_byte(self, sock: socket.socket) -> bytes | None:
        """Read one data byte, transparently handling IAC sequences."""
        while True:
            byte = sock.recv(1)
            if not byte:
                return None
            if byte[0] != IAC:
                return byte
            # IAC handling
            cmd = sock.recv(1)
            if not cmd:
                return None
            if cmd[0] == IAC:  # IAC IAC → literal 0xFF
                return b"\xff"
            if cmd[0] in (WILL, WONT, DO, DONT):  # 3-byte negotiation
                opt = sock.recv(1)
                if opt:
                    self._handle_negotiation(sock, cmd[0], opt[0])
                continue
            if cmd[0] == SB:  # Subnegotiation → skip until IAC SE
                self._skip_subnegotiation(sock)
                continue
            continue  # Other IAC commands (NOP, GA) → skip

    def _drain_pending_input(self, sock: socket.socket) -> None:
        """Drain bytes the client already sent, answering IAC sequences.

        Used after the initial negotiation window (answering queued IAC
        responses) and right before abandoning a connection (authentication
        failure / disconnect during authentication). The abandonment calls
        matter because closing a socket whose receive buffer still holds
        unread data makes TCP send RST instead of FIN (RFC 2525 2.17); on
        Windows an RST also discards data the client has not read yet, so
        anything we just sent (e.g. ``Authentication failed.``) silently
        disappears (#268). A short blocking timeout is used instead of
        non-blocking mode so that multi-byte IAC sequences split across TCP
        segments are received completely rather than raising mid-sequence;
        ``_DRAIN_TOTAL_BUDGET`` additionally bounds a client that keeps
        sending data bytes (see the constant's scope note for what it does
        NOT bound). The socket's original timeout is restored on exit.
        """
        original_timeout = sock.gettimeout()
        deadline = time.monotonic() + _DRAIN_TOTAL_BUDGET
        sock.settimeout(_IAC_DRAIN_TIMEOUT)
        try:
            while time.monotonic() < deadline:
                if self._recv_byte(sock) is None:
                    break  # EOF — client disconnected
        except TimeoutError:
            pass  # No more data available — expected
        finally:
            sock.settimeout(original_timeout)

    def _handle_negotiation(self, sock: socket.socket, cmd: int, opt: int) -> None:
        """Respond to a Telnet negotiation command."""
        if cmd == DO:
            if opt not in (SGA, ECHO):
                sock.sendall(bytes([IAC, WONT, opt]))  # Refuse unsupported
        elif cmd == WILL:
            if opt == NAWS:
                sock.sendall(bytes([IAC, DO, opt]))  # Accept NAWS
            else:
                sock.sendall(bytes([IAC, DONT, opt]))  # Refuse others
        # DONT, WONT → no response needed (already off)

    def _skip_subnegotiation(self, sock: socket.socket) -> None:
        """Skip subnegotiation data until IAC SE, handling IAC IAC escapes."""
        while True:
            byte = sock.recv(1)
            if not byte:
                return  # EOF → silently return (disconnect detected upstream)
            if byte[0] == IAC:
                next_byte = sock.recv(1)
                if not next_byte:
                    return  # EOF
                if next_byte[0] == SE:
                    return  # Normal end of subnegotiation
                # IAC IAC → escaped 0xFF in SB data, ignore and continue
                # IAC + other → protocol violation, tolerate and continue
                continue

    # ------------------------------------------------------------------
    # Authentication
    # ------------------------------------------------------------------

    def _authenticate(self, sock: socket.socket) -> tuple[bool, bool]:
        """
        Perform username/password authentication over the Telnet connection.

        Thin wrapper around tap_bridge.interactive_login supplying the
        Telnet prompts; the interaction itself is shared with SSH channel
        login. The trailing LF/NUL after a CR is no longer consumed with a
        blocking read — it is reported via skip_lf and must be forwarded to
        client_to_shell_tap (initial_skip_lf), matching SSH (U3).

        :param sock: client socket
        :return: (authenticated, skip_lf)
        """
        return interactive_login(
            TelnetSocketAdapter(sock, self),
            self.username,
            self.password,
            user_prompt=b"Username: ",
            pass_prompt=b"Password: ",
        )

    # ------------------------------------------------------------------
    # Watchdog
    # ------------------------------------------------------------------

    def watchdog(
        self,
        is_running: threading.Event,
        run_srv: threading.Event,
        shell: Any,
    ) -> None:
        """Monitor server liveness and ensure shell stops on disconnect.

        The loop exits when either ``run_srv`` is cleared (client disconnect
        detected by a tap function) or ``is_running`` is cleared (server-wide
        shutdown).  In both cases ``shell.stop()`` must be called so that
        ``CMDShell.cmdloop()`` unblocks and ``connection_function`` can return.
        """
        while run_srv.is_set():
            if not is_running.is_set():
                break
            time.sleep(min(self.watchdog_interval, SHUTDOWN_IO_TIMEOUT))
        # Always stop the shell — whether run_srv or is_running caused the exit.
        shell.stop()

    # ------------------------------------------------------------------
    # Connection handler
    # ------------------------------------------------------------------

    def connection_function(self, client: socket.socket, is_running: threading.Event) -> None:
        shell_replied_event = threading.Event()
        run_srv = threading.Event()
        run_srv.set()

        try:
            client.settimeout(self.timeout)

            # Initiate Telnet negotiation: character-at-a-time mode
            client.sendall(bytes([IAC, WILL, SGA]))
            client.sendall(bytes([IAC, WILL, ECHO]))

            # Give the client a moment to send initial IAC responses,
            # then drain them using _recv_byte so that negotiation commands
            # (e.g. DO SGA, DO ECHO, WILL NAWS) are properly answered via
            # _handle_negotiation instead of being silently discarded.
            time.sleep(0.1)
            self._drain_pending_input(client)

            # Send banner
            if self.banner:
                client.sendall((self.banner + "\r\n").encode("utf-8"))

            # Authenticate
            try:
                auth_ok, skip_lf = self._authenticate(client)
            except (TimeoutError, OSError):
                log.debug("Client disconnected during authentication")
                # Same FIN-not-RST treatment as the auth-failure branch
                # below: a server-side timeout can leave client bytes
                # pending, and closing over them would RST (#268 review).
                with contextlib.suppress(OSError):
                    self._drain_pending_input(client)
                return
            if not auth_ok:
                log.warning("Telnet authentication failed, closing connection")
                with contextlib.suppress(OSError):
                    client.sendall(b"Authentication failed.\r\n")
                # Consume the input left pending on the failure path (the
                # LF/NUL that read_line's skip_lf defers from the password
                # line's CR is forwarded to client_to_shell_tap only on
                # success) — otherwise close() RSTs and the failure message
                # never reaches Windows clients (#268).
                with contextlib.suppress(OSError):
                    self._drain_pending_input(client)
                return

            # Create stdio for the shell
            shell_stdin, shell_stdout = TapIO(run_srv), TapIO(run_srv)

            # Bridge the socket and the shell through the shared tap pair
            transport_adapter = TelnetSocketAdapter(client, self)

            # Start client→shell tap thread (skip_lf forwards the pending
            # LF/NUL from the password line's CR — U3, matches SSH)
            client_to_shell_tapper = threading.Thread(
                target=client_to_shell_tap,
                args=(transport_adapter, shell_stdin, shell_replied_event, run_srv),
                kwargs={"initial_skip_lf": skip_lf, "shell_stdout": shell_stdout},
                daemon=True,
            )
            client_to_shell_tapper.start()

            # Start shell→client tap thread
            shell_to_client_tapper = threading.Thread(
                target=shell_to_client_tap,
                args=(transport_adapter, shell_stdout, shell_replied_event, run_srv),
                daemon=True,
            )
            shell_to_client_tapper.start()

            # Create the client shell
            client_shell = self.shell(
                stdin=shell_stdin,
                stdout=shell_stdout,
                nos=self.nos,
                nos_inventory_config=self.nos_inventory_config,
                is_running=is_running,
                **self.shell_configuration,
            )

            # Start watchdog thread
            watchdog_thread = threading.Thread(
                target=self.watchdog,
                args=(is_running, run_srv, client_shell),
                daemon=True,
            )
            watchdog_thread.start()

            # Block until shell exits
            client_shell.start()
            log.debug("TelnetServer.connection_function stopped shell thread")

        finally:
            # Stop all server threads
            run_srv.clear()
            log.debug("TelnetServer.connection_function stopped server threads")

            with contextlib.suppress(OSError):
                client.close()
            log.debug("TelnetServer.connection_function closed socket")

watchdog(is_running, run_srv, shell)

Monitor server liveness and ensure shell stops on disconnect.

The loop exits when either run_srv is cleared (client disconnect detected by a tap function) or is_running is cleared (server-wide shutdown). In both cases shell.stop() must be called so that CMDShell.cmdloop() unblocks and connection_function can return.

Source code in simnos/plugins/servers/telnet_server.py

def watchdog(
    self,
    is_running: threading.Event,
    run_srv: threading.Event,
    shell: Any,
) -> None:
    """Monitor server liveness and ensure shell stops on disconnect.

    The loop exits when either ``run_srv`` is cleared (client disconnect
    detected by a tap function) or ``is_running`` is cleared (server-wide
    shutdown).  In both cases ``shell.stop()`` must be called so that
    ``CMDShell.cmdloop()`` unblocks and ``connection_function`` can return.
    """
    while run_srv.is_set():
        if not is_running.is_set():
            break
        time.sleep(min(self.watchdog_interval, SHUTDOWN_IO_TIMEOUT))
    # Always stop the shell — whether run_srv or is_running caused the exit.
    shell.stop()

Internal

TapIO

Thread-safe I/O bridge shared by both SSH and Telnet server plugins. This is an internal helper, not a public API.

Bases: StringIO

Class to implement StringIO subclass but with blocking readline method and a deque to buffer lines on write.

Uses collections.deque for thread-safe, O(1) append/pop operations (CPython's GIL guarantees atomicity for deque append/pop).

A threading.Condition is used to wake readline() immediately when write() adds data, eliminating the polling delay that caused intermittent empty output in netmiko send_command() (#87).

Source code in simnos/plugins/servers/tap_io.py

class TapIO(io.StringIO):
    """
    Class to implement StringIO subclass but with blocking readline method
    and a deque to buffer lines on write.

    Uses ``collections.deque`` for thread-safe, O(1) append/pop operations
    (CPython's GIL guarantees atomicity for deque ``append``/``pop``).

    A ``threading.Condition`` is used to wake ``readline()`` immediately
    when ``write()`` adds data, eliminating the polling delay that caused
    intermittent empty output in netmiko ``send_command()`` (#87).
    """

    def __init__(self, run_srv: threading.Event, initial_value: str = "", newline: str = "\n"):
        self.lines: deque[str] = deque()
        self.run_srv: threading.Event = run_srv
        self._cond: threading.Condition = threading.Condition()
        super().__init__(initial_value, newline)

    def readline(self, size: int | None = -1) -> str:
        """Block until a line is available or the server shuts down.

        The ``size`` argument is accepted for compatibility with
        ``io.IOBase.readline`` but ignored; this stream returns complete
        lines from the internal queue regardless of byte budget.
        """
        del size  # parent-class signature compatibility only
        with self._cond:
            while self.run_srv.is_set():
                if self.lines:
                    return self.lines.pop()
                self._cond.wait(timeout=0.1)
        if self.lines:
            return self.lines.pop()
        return ""

    def drain(self) -> list[str]:
        """Pop all buffered lines without blocking.

        Returns a list in FIFO order (oldest first).
        """
        items: list[str] = []
        while self.lines:
            items.append(self.lines.pop())
        return items

    def write(self, value: str):
        """Append *value* to the buffer and wake any blocked ``readline()``."""
        self.lines.appendleft(value)
        with self._cond:
            self._cond.notify()

drain()

Pop all buffered lines without blocking.

Returns a list in FIFO order (oldest first).

Source code in simnos/plugins/servers/tap_io.py

def drain(self) -> list[str]:
    """Pop all buffered lines without blocking.

    Returns a list in FIFO order (oldest first).
    """
    items: list[str] = []
    while self.lines:
        items.append(self.lines.pop())
    return items

readline(size=-1)

Block until a line is available or the server shuts down.

The size argument is accepted for compatibility with io.IOBase.readline but ignored; this stream returns complete lines from the internal queue regardless of byte budget.

Source code in simnos/plugins/servers/tap_io.py

def readline(self, size: int | None = -1) -> str:
    """Block until a line is available or the server shuts down.

    The ``size`` argument is accepted for compatibility with
    ``io.IOBase.readline`` but ignored; this stream returns complete
    lines from the internal queue regardless of byte budget.
    """
    del size  # parent-class signature compatibility only
    with self._cond:
        while self.run_srv.is_set():
            if self.lines:
                return self.lines.pop()
            self._cond.wait(timeout=0.1)
    if self.lines:
        return self.lines.pop()
    return ""

write(value)

Append value to the buffer and wake any blocked readline().

Source code in simnos/plugins/servers/tap_io.py

def write(self, value: str):
    """Append *value* to the buffer and wake any blocked ``readline()``."""
    self.lines.appendleft(value)
    with self._cond:
        self._cond.notify()

Shell Plugins

Shell plugins act as plumbing between server plugins and NOS plugins, connecting them together.

CMDShell

Bases: Cmd

Custom shell class to interact with NOS.

Source code in simnos/plugins/shell/cmd_shell.py

class CMDShell(Cmd):
    """
    Custom shell class to interact with NOS.
    """

    use_rawinput = False

    def __init__(
        self,
        stdin,
        stdout,
        nos,
        nos_inventory_config,
        base_prompt,
        is_running,
        intro="Custom SSH Shell",
        ruler="",
        completekey="tab",
        newline="\r\n",
    ):
        self.nos: Nos = nos
        self.ruler = ruler
        self.intro = intro
        self.base_prompt = base_prompt
        self.newline = newline
        # Lenient: a malformed initial_prompt template must not kill every
        # connection to this host; fall back to the raw template and log.
        formatted = self._safe_format(nos.initial_prompt, where="initial_prompt")
        self.prompt = formatted if formatted is not None else nos.initial_prompt
        self.is_running = is_running

        # form commands. Inventory-defined commands are the one source
        # that does not pass through the `Nos` load path, so their
        # `prompt` is normalized here (str -> [str], on our own deepcopy)
        # to uphold the lists-only read-side contract (#244 / D3).
        self.commands = {
            **copy.deepcopy(BASIC_COMMANDS),
            **copy.deepcopy(nos.commands or {}),
            **Nos.normalize_command_prompts(copy.deepcopy(nos_inventory_config.get("commands", {}))),
        }
        # call the base constructor of cmd.Cmd, with our own stdin and stdout
        super().__init__(
            completekey=completekey,
            stdin=stdin,
            stdout=stdout,
        )

    def start(self):
        """Method to start the shell"""
        self.cmdloop()

    def stop(self):
        """Method to stop the shell"""
        self.stdin.write("exit" + self.newline)

    def writeline(self, value):
        """Method to write a line to stdout with newline at the end"""
        for line in str(value).splitlines():
            self.stdout.write(line + self.newline)

    def do_EOF(self, line):
        """Handle EOF from readline — exit the shell gracefully."""
        return True

    def emptyline(self):
        """This method to do nothing if empty line entered"""

    def reload_commands(self, changed_files: list):
        """Method to reload commands

        Lenient per file: hot reload is a dev feature and may observe a
        half-written or malformed plugin file (e.g. an editor's partial
        save, or a file that vanished after detection). One broken file
        must not kill the SSH session nor block reloading the remaining
        files — log and retry on the next change.
        """
        for file in changed_files:
            try:
                self.nos.from_file(file)
            except Exception:
                # Broad except, like `default()`: any plugin error must not
                # crash the session. The traceback goes to the log so a
                # genuine plugin bug surfacing here stays diagnosable.
                log.error("shell '%s' failed to hot-reload %r\n%s", self.base_prompt, file, traceback.format_exc())
                continue
            self.commands.update(self.nos.commands)

    def precmd(self, line):
        """Method to return line before processing the command"""
        if os.environ.get("SIMNOS_RELOAD_COMMANDS"):
            changed_files = get_files_changed(nos.__path__[0])
            if changed_files:
                log.debug("Reloading... Files changed: %s", changed_files)
                self.reload_commands(changed_files)
        return line

    def postcmd(self, stop, line):
        """Method to return stop value to stop the shell"""
        return stop

    def do_help(self, arg):
        """Method to return help for commands"""
        lines = {}  # dict of {cmd: cmd_help}
        width = 0  # record longest command width for padding
        # form help for all commands
        for cmd, cmd_data in self.commands.items():
            # skip special commands
            if cmd.startswith("_") and cmd.endswith("_"):
                continue
            # skip commands that does not match current prompt
            if not self._check_prompt(cmd_data.get("prompt"), command=cmd):
                continue
            lines[cmd] = cmd_data.get("help", "")
            width = max(width, len(cmd))
        # form help lines
        help_msg = []
        for k, v in lines.items():
            padding = " " * (width - len(k)) + "  "
            help_msg.append(f"{k}{padding}{v}")
        self.writeline(self.newline.join(help_msg))

    def _safe_format(self, template: str, *, where: str) -> str | None:
        """Format `template` with `base_prompt`; return None on failure.

        The runtime shell is intentionally lenient: yaml templating errors
        are logged but never crash the session nor leak tracebacks to the
        wire. The build-time counterpart `tasks.render_template` shares the
        `FORMAT_ERRORS` catch set but raises `RuntimeError` — and is
        additionally strict about unsupported constructs that would render
        fine (e.g. `{base_prompt!r}`). Yaml authors may use only
        `{base_prompt}` substitution and `{{` / `}}` escapes; see
        docs/development/creating_new_platforms.md.

        :param template: format template string from yaml / plugin data
        :param where: caller context for the error log; include the command
            name when available (e.g. ``f"output for command {line!r}"``)
        """
        try:
            return template.format(base_prompt=self.base_prompt)
        except FORMAT_ERRORS as e:
            log.error(
                "shell '%s' error formatting %s %r: %r",
                self.base_prompt,
                where,
                template,
                e,
            )
            return None

    def _check_prompt(self, prompt_: list[str] | None, command: str = ""):
        """
        Helper method to check if prompt_ matches current prompt

        :param prompt_: (list of strings, or None) prompt to check — every
            load path normalizes a bare-str authoring form to a list
            before commit (#244 / D3), so no str branch is needed here
        :param command: command name for the error log; callers without it
            keep working (the log just omits the command context)
        """
        # prompt_ is None if no 'prompt' key defined for command
        if prompt_ is None:
            return True
        candidates = prompt_
        where = f"prompt for command {command!r}" if command else "prompt"
        # A broken candidate is just a non-match; the remaining candidates
        # are still evaluated independently.
        for candidate in candidates:
            formatted = self._safe_format(candidate, where=where)
            if formatted is not None and self.prompt == formatted:
                return True
        return False

    def _apply_new_prompt(self, template: str, command: str) -> None:
        """Transition the prompt; a broken template keeps the current one.

        Shared by the callable-dict and cmd_data `new_prompt` paths of
        `default()`: a format failure means no prompt transition (the
        session stays on the current prompt); see `_safe_format`.
        """
        new_prompt = self._safe_format(template, where=f"new_prompt for command {command!r}")
        if new_prompt is not None:
            self.prompt = new_prompt

    def _resolve_command(self, command: str) -> dict | None:
        """Return the merged cmd_data for `command`, or None if unknown.

        Alias resolution happens here too: a missing alias target is the
        same lenient unknown-command path as a missing command (both used
        to be one broad `except KeyError`) — the caller answers with the
        `_default_` output, never with the handler-crash response.
        """
        try:
            cmd_data = self.commands[command]
            if "alias" in cmd_data:
                cmd_data = {**self.commands[cmd_data["alias"]], **cmd_data}
        except KeyError:
            log.error("shell.default '%s' command '%s' not found", self.base_prompt, [command])
            return None
        return cmd_data

    def _invoke_callable(self, func: CommandHandler, command: str) -> CommandResult:
        """Invoke a command handler and normalize its return to CommandResult.

        A plain str (or None) return is sugar for `{"output": <value>}`;
        see `simnos.core.command_contract`. This is normalization, not
        validation: a contract-breaking return (list / int / ...) is
        wrapped and flows through the lenient output path like today —
        contract violations are caught statically (Protocol) and by the
        e2e callable sweep, not at runtime on the hot path.
        """
        ret = func(
            self.nos.device,
            base_prompt=self.base_prompt,
            current_prompt=self.prompt,
            command=command,
        )
        if isinstance(ret, dict):
            return ret
        # Declare the wrapped value as str | None for the type checker
        # (it cannot narrow the TypedDict member out of the union by
        # isinstance). The declaration matches the *contract*, not a
        # runtime guarantee — a contract-breaking return (list / int)
        # is wrapped as-is and absorbed by the lenient output path.
        return {"output": cast("str | None", ret)}

    def _render_output(self, ret, command: str, *, format_output: bool) -> None:
        """Write `ret` to the client; only yaml-static output is formatted.

        `ret` is untyped on purpose: the lenient path also carries
        contract-breaking handler returns (see `_invoke_callable`), which
        `writeline`'s `str(value)` absorbs. The caller must not pass
        None though — `default()` guards with `if ret is not None`
        ("write nothing"), this helper always writes.

        Callable output is passed through verbatim (`format_output=False`):
        handlers receive `base_prompt` as an argument and format
        themselves (see `CommandHandler`), so a second `.format()` here
        would only mis-render device output containing literal braces or
        an accidental `{base_prompt}` (#241 / D-b). For yaml-static
        output, a format failure falls back to the raw template
        (information beats dropping the whole output); lenient policy in
        `_safe_format`.
        """
        if not format_output:
            self.writeline(ret)
            return
        formatted = self._safe_format(ret, where=f"output for command {command!r}")
        self.writeline(formatted if formatted is not None else ret)

    def default(self, line):
        """Dispatch `line`: resolve -> prompt check -> invoke -> render.

        The exception boundary is the `_invoke_callable` block only:
        resolve / alias / prompt check / new_prompt never raise (KeyError
        degrades inside `_resolve_command`, format errors are caught
        inside `_safe_format`), so `HANDLER_ERROR_OUTPUT` is structurally
        guaranteed to mean "a command handler crashed" and nothing else.
        """
        log.debug("shell.default '%s' running command '%s'", self.base_prompt, [line])
        from_callable = False
        cmd_data = self._resolve_command(line)
        if cmd_data is not None and self._check_prompt(cmd_data.get("prompt"), command=line):
            if cmd_data.get("exit"):
                return True
            ret = cmd_data.get("output")
        else:
            if cmd_data is not None:
                log.warning(
                    "'%s' command prompt '%s' not matching current prompt '%s'",
                    line,
                    # Always a list here: a mismatch requires a non-None,
                    # normalized prompt (#244 / D3).
                    ", ".join(cmd_data.get("prompt", [])),
                    self.prompt,
                )
            # Unknown command and prompt mismatch both answer with the
            # `_default_` output — a silent shell would make clients
            # (e.g. Netmiko) wait for a timeout instead.
            ret = self.commands["_default_"]["output"]
            cmd_data = None  # the `_default_` answer never applies cmd_data's new_prompt
        if callable(ret):
            from_callable = True
            try:
                result = self._invoke_callable(ret, line)
            except Exception:
                # Same shape as the hot-reload guard (#232): full traceback
                # to the log, the session survives, and the client gets a
                # real-NOS-style one-liner instead of a Python traceback.
                log.error(
                    "shell '%s' command %r handler crashed\n%s",
                    self.base_prompt,
                    line,
                    traceback.format_exc(),
                )
                result = {"output": HANDLER_ERROR_OUTPUT}
            if "new_prompt" in result:
                self._apply_new_prompt(result["new_prompt"], line)
            if result.get("exit"):
                return True
            ret = result.get("output")
        if cmd_data is not None and "new_prompt" in cmd_data:
            self._apply_new_prompt(cmd_data["new_prompt"], line)
        if not self.is_running.is_set():
            return True
        if ret is not None:
            self._render_output(ret, line, format_output=not from_callable)
        return False

default(line)

Dispatch line: resolve -> prompt check -> invoke -> render.

The exception boundary is the _invoke_callable block only: resolve / alias / prompt check / new_prompt never raise (KeyError degrades inside _resolve_command, format errors are caught inside _safe_format), so HANDLER_ERROR_OUTPUT is structurally guaranteed to mean "a command handler crashed" and nothing else.

Source code in simnos/plugins/shell/cmd_shell.py

def default(self, line):
    """Dispatch `line`: resolve -> prompt check -> invoke -> render.

    The exception boundary is the `_invoke_callable` block only:
    resolve / alias / prompt check / new_prompt never raise (KeyError
    degrades inside `_resolve_command`, format errors are caught
    inside `_safe_format`), so `HANDLER_ERROR_OUTPUT` is structurally
    guaranteed to mean "a command handler crashed" and nothing else.
    """
    log.debug("shell.default '%s' running command '%s'", self.base_prompt, [line])
    from_callable = False
    cmd_data = self._resolve_command(line)
    if cmd_data is not None and self._check_prompt(cmd_data.get("prompt"), command=line):
        if cmd_data.get("exit"):
            return True
        ret = cmd_data.get("output")
    else:
        if cmd_data is not None:
            log.warning(
                "'%s' command prompt '%s' not matching current prompt '%s'",
                line,
                # Always a list here: a mismatch requires a non-None,
                # normalized prompt (#244 / D3).
                ", ".join(cmd_data.get("prompt", [])),
                self.prompt,
            )
        # Unknown command and prompt mismatch both answer with the
        # `_default_` output — a silent shell would make clients
        # (e.g. Netmiko) wait for a timeout instead.
        ret = self.commands["_default_"]["output"]
        cmd_data = None  # the `_default_` answer never applies cmd_data's new_prompt
    if callable(ret):
        from_callable = True
        try:
            result = self._invoke_callable(ret, line)
        except Exception:
            # Same shape as the hot-reload guard (#232): full traceback
            # to the log, the session survives, and the client gets a
            # real-NOS-style one-liner instead of a Python traceback.
            log.error(
                "shell '%s' command %r handler crashed\n%s",
                self.base_prompt,
                line,
                traceback.format_exc(),
            )
            result = {"output": HANDLER_ERROR_OUTPUT}
        if "new_prompt" in result:
            self._apply_new_prompt(result["new_prompt"], line)
        if result.get("exit"):
            return True
        ret = result.get("output")
    if cmd_data is not None and "new_prompt" in cmd_data:
        self._apply_new_prompt(cmd_data["new_prompt"], line)
    if not self.is_running.is_set():
        return True
    if ret is not None:
        self._render_output(ret, line, format_output=not from_callable)
    return False

do_EOF(line)

Handle EOF from readline — exit the shell gracefully.

Source code in simnos/plugins/shell/cmd_shell.py

def do_EOF(self, line):
    """Handle EOF from readline — exit the shell gracefully."""
    return True

do_help(arg)

Method to return help for commands

Source code in simnos/plugins/shell/cmd_shell.py

def do_help(self, arg):
    """Method to return help for commands"""
    lines = {}  # dict of {cmd: cmd_help}
    width = 0  # record longest command width for padding
    # form help for all commands
    for cmd, cmd_data in self.commands.items():
        # skip special commands
        if cmd.startswith("_") and cmd.endswith("_"):
            continue
        # skip commands that does not match current prompt
        if not self._check_prompt(cmd_data.get("prompt"), command=cmd):
            continue
        lines[cmd] = cmd_data.get("help", "")
        width = max(width, len(cmd))
    # form help lines
    help_msg = []
    for k, v in lines.items():
        padding = " " * (width - len(k)) + "  "
        help_msg.append(f"{k}{padding}{v}")
    self.writeline(self.newline.join(help_msg))

emptyline()

This method to do nothing if empty line entered

Source code in simnos/plugins/shell/cmd_shell.py

def emptyline(self):
    """This method to do nothing if empty line entered"""

postcmd(stop, line)

Method to return stop value to stop the shell

Source code in simnos/plugins/shell/cmd_shell.py

def postcmd(self, stop, line):
    """Method to return stop value to stop the shell"""
    return stop

precmd(line)

Method to return line before processing the command

Source code in simnos/plugins/shell/cmd_shell.py

def precmd(self, line):
    """Method to return line before processing the command"""
    if os.environ.get("SIMNOS_RELOAD_COMMANDS"):
        changed_files = get_files_changed(nos.__path__[0])
        if changed_files:
            log.debug("Reloading... Files changed: %s", changed_files)
            self.reload_commands(changed_files)
    return line

reload_commands(changed_files)

Method to reload commands

Lenient per file: hot reload is a dev feature and may observe a half-written or malformed plugin file (e.g. an editor's partial save, or a file that vanished after detection). One broken file must not kill the SSH session nor block reloading the remaining files — log and retry on the next change.

Source code in simnos/plugins/shell/cmd_shell.py

def reload_commands(self, changed_files: list):
    """Method to reload commands

    Lenient per file: hot reload is a dev feature and may observe a
    half-written or malformed plugin file (e.g. an editor's partial
    save, or a file that vanished after detection). One broken file
    must not kill the SSH session nor block reloading the remaining
    files — log and retry on the next change.
    """
    for file in changed_files:
        try:
            self.nos.from_file(file)
        except Exception:
            # Broad except, like `default()`: any plugin error must not
            # crash the session. The traceback goes to the log so a
            # genuine plugin bug surfacing here stays diagnosable.
            log.error("shell '%s' failed to hot-reload %r\n%s", self.base_prompt, file, traceback.format_exc())
            continue
        self.commands.update(self.nos.commands)

start()

Method to start the shell

Source code in simnos/plugins/shell/cmd_shell.py

def start(self):
    """Method to start the shell"""
    self.cmdloop()

stop()

Method to stop the shell

Source code in simnos/plugins/shell/cmd_shell.py

def stop(self):
    """Method to stop the shell"""
    self.stdin.write("exit" + self.newline)

writeline(value)

Method to write a line to stdout with newline at the end

Source code in simnos/plugins/shell/cmd_shell.py

def writeline(self, value):
    """Method to write a line to stdout with newline at the end"""
    for line in str(value).splitlines():
        self.stdout.write(line + self.newline)

Tape Plugins

Idea - Tape Plugins will allow to record interactions with real devices and build NOS plugins automatically using gathered data.