ExecSequence

An ExecSequence is a declarative, serializable definition of a terminal interaction. A single ExecSequence can include various read and write operations combined in any order. For instance:

from radkit_client import ExecSequence

seq = ExecSequence().start_capture().write("show version\r").readuntil_prompt()

The above ExecSequence starts by enabling capturing, so that we get the captured output as a string upon completion. Then the ExecSequence writes show version\r to the standard input stream of the terminal, and finally it reads from the output stream until we discover a prompt.

An ExecSequence is serializable, but it does not rely on Python pickling, which means that it can safely be sent to a different machine for execution without any risk of remote code execution through Python instructions (no Python code is sent over the wire, only serialized objects).

Usage on InteractiveConnection

In RADKit Client, we can create an ExecSequence and then have it executed on an open terminal connection.

We have the choice to perform the execution locally in the Client, or to send the ExecSequence to the Service for remote execution. There are some advantages when executing remotely:

• The ExecSequence will have access to the inventory with all the device credentials. So, if the ExecSequence includes instructions that require credentials (like the enable-privileged-mode procedure), those credentials can be leveraged without being made available to the Client.

• The captured data can be uploaded elsewhere without transferring it to the Client first. This is particularly useful for large amounts of data that must eventually be uploaded to the TAC case notes.

These are the methods for executing an ExecSequence on a terminal connection:

• radkit_client.sync.InteractiveConnection.run_exec_sequence()

• radkit_client.sync.InteractiveConnection.run_remote_exec_sequence()

• radkit_client.sync.InteractiveConnection.run_remote_exec_sequence_as_connection()

# ExecSequence gets imported automatically into the REPL scope,
# but it can also be imported from radkit_client in a script.

# Define the ExecSequence
seq = (
    ExecSequence()
        .readuntil_timeout(timeout=1)
        .start_capture()
        .write("show version\r")
        .readuntil_prompt()
)

# Open a terminal connection to a device
terminal = service.inventory["some-device"].terminal()

# Run the ExecSequence in RADKit Client and capture the result in a variable
result = terminal.run_exec_sequence(seq)

# Send the ExecSequence to RADKit Service for remote execution
result = terminal.run_remote_exec_sequence(seq)

# Attach to the terminal for some further manual interaction
# (type ~. after hitting Enter to detach)
terminal.attach()

The captured data will be returned from the run_exec_sequence() or run_remote_exec_sequence() terminal methods, unless it’s uploaded elsewhere (see below).

Uploading to CXD

When execution takes place on the Service using run_remote_exec_sequence(), the captured data can be uploaded elsewhere using the upload_to parameter, without streaming the data first to the Client (see Upload parameters for more information).

# Open a terminal connection to a device
terminal = service.inventory["some-device"].terminal()

# Maybe attach and do some manual interaction first
# (type ~. after hitting Enter to detach)
terminal.attach()

# Define the ExecSequence
seq = ExecSequence().start_capture().exec("show version")

# Run the ExecSequence remotely on the Service; the captured data will be
# streamed directly to the upload destination, and won't be sent to the Client
terminal.run_remote_exec_sequence(seq, upload_to=cxd(...))

# Maybe attach again for further manual interaction
terminal.attach()

See CX Drive (CXD) for more information about uploading to CXD.

ExecSequence methods

The ExecSequence methods largely correspond to the read/write methods of an InteractiveConnection, plus several others.

The following methods are available (for details, click the method name or refer to the Client API reference):

read()

Reads up to a given number of bytes, or until EOF.

readexactly()

Reads exactly a given number of bytes.

readuntil()

Reads until a given substring is found.

readuntil_regex()

Reads until a given regex matches.

readline()

Reads until the end of the current line.

readuntil_timeout()

Reads until no more data is coming in for a given amount of time.

readuntil_prompt()

Reads until a prompt is detected (using automatic prompt detection).

write()

Writes data.

write_eof()

Terminates the input stream by writing EOF.

exec()

Executes a command and waits for the next prompt.

set_terminal_size()

Sets the terminal size (for connections that support it).

sleep()

Sleeps for a given amount of time.

start_capture()

Starts capturing output.

end_capture()

Stops capturing output.

capture()

Sends a given string to the capture stream (for debugging/annotating).

call_procedure()

Calls a hardcoded procedure (see below).

enter_privileged_mode()

Enters privileged mode on the device (if available/configured).

include()

Includes another ExecSequence object.

Procedures

An ExecSequence can refer to “procedures” i.e. more complex terminal interactions with a given name:

# Perform the "maglev" login sequence on Catalyst Center
seq = ExecSequence().start_capture().call_procedure("maglev-login")

These procedures have to be available in the Client code that is running locally, or the Service that is running remotely, depending on where the ExecSequence is executed.

The following procedures are available:

enable-privileged-mode (added in 1.6.0)

Issues the enable command and expects a password prompt or a privileged prompt (#). The enable password is configured separately in the Terminal device properties on the Service.

disable-pager (added in 1.6.0)

Sends the appropriate command(s) to disable the pager on the device (for example: term len 0, term wid 0 on IOS XE/XR; term pager 0 on ASA; etc.)

maglev-login (added in 1.6.0)

Performs maglev login on a Catalyst Center. Provides the username and password configured in the HTTP device properties on the Service.

maglev-rca (added in 1.6.0)

Performs rca on a Catalyst Center. Requires the username and password configured in the HTTP device properties on the Service, as well as the password configured under Terminal device properties.

Composing multiple ExecSequence objects together

Thanks to the include() method, ExecSequence objects can be nested. This makes it possible to build a library of reusable ExecSequence recipes and include them as part of other ExecSequence instances later on.

# Create two reusable ExecSequence objects
show_version_seq = (
    ExecSequence()
        .readuntil_timeout(timeout=1)
        .write("show version\r")
        .readuntil_prompt()
)
show_clock_seq = (
    ExecSequence()
        .readuntil_timeout(timeout=1)
        .write("show clock\r")
        .readuntil_prompt()
)

# Use them as part of the main ExecSequence
main_seq = (
    ExecSequence()
        .start_capture()
        .include(show_version_seq)
        .include(show_clock_seq)
)
result = terminal.run_remote_exec_sequence(main_seq)

It is often useful to define a function that produces an ExecSequence to avoid the repetition:

def run_show_command(command: str) -> ExecSequence:
    " Creates an ExecSequence that executes this 'show' command. "
        return (
            ExecSequence()
            .readuntil_timeout(timeout=1)
            .write(f"show {command}\r")
            .readuntil_prompt()
        )

main_seq = (
    ExecSequence()
    .start_capture()
    .include(run_show_command("clock"))
    .include(run_show_command("version"))
)
result = terminal.run_remote_exec_sequence(main_seq)

Note

The code shown above is only meant as a trivial example. This particular result can of course also be achieved using exec() instead:

ExecSequence().start_capture().exec(...)