Built-in Plugins API

NOS Plugins#

NOS plugins are at the heart of FakeNOS, 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 fakenos/plugins/nos/platforms_py/cisco_ios.py

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

    hola = "Hello test"

    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 fakenos/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 fakenos/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 fakenos/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 fakenos/plugins/servers/ssh_server_paramiko.py

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

    # pylint: disable=too-many-instance-attributes
    # pylint: disable=too-many-arguments
    def __init__(
        self,
        shell: type,
        nos: Nos,
        nos_inventory_config: Dict,
        port: int,
        username: str,
        password: str,
        ssh_key_file: paramiko.rsakey.RSAKey = None,
        ssh_key_file_password: str = None,
        ssh_banner: str = "FakeNOS Paramiko SSH Server",
        shell_configuration: Dict = None,
        address: str = "127.0.0.1",
        timeout: int = 1,
        watchdog_interval: int = 1,
    ):
        super().__init__()

        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.port: int = port
        self.address: str = address
        self.timeout: int = timeout
        self.watchdog_interval: int = watchdog_interval

        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:
            self._ssh_server_key: paramiko.rsakey.RSAKey = paramiko.RSAKey(file_obj=io.StringIO(DEFAULT_SSH_KEY))

    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 connection_function(self, client: socket.socket, is_running: threading.Event):
        shell_replied_event = threading.Event()
        run_srv = threading.Event()
        run_srv.set()

        # create the SSH transport object
        session = paramiko.Transport(client)
        session.add_server_key(self._ssh_server_key)

        # create the server
        server = ParamikoSshServerInterface(
            ssh_banner=self.ssh_banner,
            username=self.username,
            password=self.password,
        )

        # start the SSH server
        session.start_server(server=server)

        # create the channel and get the stdio
        channel = session.accept()
        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 fakenos/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.