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)

Servers Plugins¶

Servers Plugins acts 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
    _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 support in server mode.
        # Result is cached at the class level so subsequent instances skip the file I/O.
        if ParamikoSshServer._moduli_loaded is None:
            ParamikoSshServer._moduli_loaded = paramiko.Transport.load_server_moduli()

    @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")
                shell.stop()
                break

            if not is_running.is_set():
                shell.stop()

            time.sleep(self.watchdog_interval)

    def _read_channel_line(self, channel, echo: bool = True) -> str:
        """
        Read a single line from the channel byte-by-byte.

        :param channel: paramiko Channel
        :param echo: if True, echo each byte back to the client
        :return: the line read (without trailing CR/LF)
        """
        channel.settimeout(self.timeout)
        buf = b""
        while True:
            try:
                byte = channel.recv(1)
            except TimeoutError:
                continue
            if byte in (b"", None):
                break
            if byte in (b"\r", b"\n"):
                if echo:
                    channel.sendall(b"\r\n")
                break
            if echo:
                channel.sendall(byte)
            buf += byte
        return buf.decode("utf-8", errors="replace")

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

        Sends "User Name:" / "Password:" prompts and validates credentials.

        :param channel: paramiko Channel
        :return: True if login succeeded, False otherwise
        """
        channel.sendall(b"\r\nUser Name:")
        username = self._read_channel_line(channel, echo=True)
        channel.sendall(b"\r\nPassword:")
        password = self._read_channel_line(channel, echo=False)
        channel.sendall(b"\r\n")

        if username == self.username and password == self.password:
            log.debug("Channel login succeeded for user %s", username)
            return True

        log.warning("Channel login failed for user %s", username)
        return False

    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 self._moduli_loaded:
            session.disabled_algorithms = _DISABLED_GEX_ALGORITHMS
        session.add_server_key(self._ssh_server_key)

        # 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
        session.start_server(server=server)

        # create the channel and get the stdio
        channel = session.accept()
        if channel is None:
            log.warning("session.accept() returned None, closing transport")
            session.close()
            return

        # 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.
        if server.auth_method_used == "none" and not self._channel_login(channel):
            log.warning("Channel login failed, closing connection")
            channel.close()
            session.close()
            return

        channel_stdio = channel.makefile("rw")

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

        # start intermediate thread to tap into
        # the channel_stdio->shell_stdin bytes stream
        channel_to_shell_tapper = threading.Thread(
            target=channel_to_shell_tap,
            args=(channel_stdio, shell_stdin, shell_replied_event, run_srv),
        )
        channel_to_shell_tapper.start()

        # start intermediate thread to tap into
        # the shell_stdout->channel_stdio bytes stream
        shell_to_channel_tapper = threading.Thread(
            target=shell_to_channel_tap,
            args=(channel_stdio, shell_stdout, shell_replied_event, run_srv),
        )
        shell_to_channel_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))
        watchdog_thread.start()

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

        # kill this server threads - watchdog, TapIO,
        # shell_to_channel_tapper and channel_to_shell_tapper
        run_srv.clear()
        log.debug("ParamikoSshServer.connection_function stopped server threads")

        # After execution continues, we can close the session
        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")
            shell.stop()
            break

        if not is_running.is_set():
            shell.stop()

        time.sleep(self.watchdog_interval)

Shell Plugins¶

Shell Plugins act as a plumbing between servers plugins and NOS plugins, gluing, connecting them together.