Reading long output on device without datadump

[Sciti]

Greetings!
With GenericDriver is it possible to scroll until '--more--' disappear?
Only one thing came into my mind right now - its to create base Driver for this purpose and send channel.send(' ') until '--more--' not in channel.read().. Looks not right.
Looks like EdgeCore ES doesn't provide any config for terminal :(

[carlmontanari]

Hey @Sciti are you sure this doesn’t work? I know folks are using scrapli and edgecore with the community driver.

To answer your question, “yes” and “no”. “No” this is not the generic drivers problem, it gives you the base setup and then it’s your responsibility to do whatever you need from there. The “yes” part: sure, you have all the tools you need because can read directly from the transport until you see whatever pattern, then write a space/enter/whatever. But I would bet money you don’t need to do this, you just need to disable paging on the device like the community platform does.

[Sciti]

Hey @Sciti are you sure this doesn’t work? I know folks are using scrapli and edgecore with the community driver.

To answer your question, “yes” and “no”. “No” this is not the generic drivers problem, it gives you the base setup and then it’s your responsibility to do whatever you need from there. The “yes” part: sure, you have all the tools you need because can read directly from the transport until you see whatever pattern, then write a space/enter/whatever. But I would bet money you don’t need to do this, you just need to disable paging on the device like the community platform does.

Yup. edgecore driver works, but as i see in device docs, it doesn't have anything related to terminal length. And thats the problem, because with edgecore community driver, it stuck on '--more--'

Tried to use something like this, but i don't fully understand right now how it's actually works..

conn.channel.write(' ')
conn.channel.write('sh mac-address-table\n')
result = conn.channel.read()
while b'---more---' in result:   # tried with and without b''
    conn.channel.write(' ')    
    result = conn.channel.read()

It's still stuck
Here's more readable manual

[denisbondar]

Hi!
Joining me in the question. There are quite a few CLI implementations that lack a way to disable pagination.
Right now I'm working with ZyXEL equipment (MES series switches) which have no commands to disable pagination or change terminal length.
And it's not the only vendor that doesn't provide a non-page output option.

My idea of working with such equipment is as follows.
Using the example of how the driver method send_command() waits for a command line prompt, implement in this method waiting for the string --more-- or any other regex string configurable through the driver configuration and sending in response to this string a space character or any other character also configurable through the driver configuration.
Thus, we will always have the whole data at the output of the send_command() method, and all pagination will be performed "under the hood".

How can this be implemented?

[denisbondar]

Kind of like how it won't work in a loop. That is, if the output is very long and you need "to press space" many times, it seems to me that this method will not work.

[Sciti]

But name of this function is exactly what i need, send input and interact...

[denisbondar]

I probably confused it with the driver's send_interactive().

[carlmontanari]

ill start with this just to be clear: scrapli will not change with respect to this. there wont be a "auto paginate" mode. so, I am 100% positive you can make it work but you're on your own for it. I suspect we can all get to a good solution here, just it wont be something that goes into scrpali (probably... never say never but probably not :P)

read_with_callback and send_interactive can probably get shoe horned into working here. I probably wouldn't use them for this use case though (maybe the callback one since it's basically a better version of send interactive and I think it could probably do it). I think I mentioned before, btu basically you're going to have to bypass all of the "normal" driver stuff and use channel methods directly -- like @denisbondar was saying you can mess with prompt and use send_input but I think you would probably have an easier time "going lower". scrapli replay would be a good thing to inspect for all this I think esp here.

[Sciti]

It'll be enough just to find some solution for this C: I'm not so good at python, but it's my first task as junior python dev and i really want to get it working. Thanks for this clues, i hope we can find something together with Denis!

[denisbondar]

@Sciti I have a Edge-Core ES3528M switch. It has the ability to set the terminal length. The following may be of some help to you:

Vty-0#terminal length ?
  <0-512>  Number of lines on screen (0 for no pausing)

Vty-0#terminal length 0

Vty-0#show system 
System Description: Edge-Core FE L2 Switch ES3528M
System OID String: 1.3.6.1.4.1.259.6.10.94

Vty-0#show version 
Unit 1
 Serial Number:           203001567
 Hardware Version:        R01
 Chip Device ID:          Marvell 98DX106-B0, 88E6095[F]
 EPLD Version:            0.07
 Number of Ports:         28
 Main Power Status:       Up
 Redundant Power Status:  Not present

Agent (Master)
 Unit ID:                 1
 Loader Version:          1.0.4.0
 Boot ROM Version:        1.2.1.2
 Operation Code Version:  1.4.20.27

[Sciti]

Nope, i tried :c

3S24-1-0#terminal length ?
% Unrecognized command.
3S24-1-0#sh system
System Description: Layer2+ Fast Ethernet Standalone Switch ES3526XA
...

[denisbondar]

I can't seem to find a way to disable pagination on TP-Link switches with the first firmware version either. These are JetStream switches (TL-SG) as well as TL-SL series switches.

There is a way to disable pagination on TP-Link switches with the second firmware version and I know it.

[Sciti]

@denisbondar Greetings, I got something working, thanks to Carl advice for read_callback! Look into read_callback, there's great example, but now another problem appeared 😓
To read full output and continue executing i wanted to make promt as second callback with complete=True, but it reads it at start and finishes. Any thoughts?

[carlmontanari]

was bored at the airport, so... this should do what you both want I think. this is sending an escape instead of a space or whatever other character(s) your device expects of course, so it just ends instead but you get the gist.

from scrapli.driver.generic.base_driver import ReadCallback
from scrapli.driver.generic.sync_driver import GenericDriver


device = {
    "host": "172.31.254.1",
    "auth_strict_key": False,
    "ssh_config_file": True,
}


class Paginator:
    def __init__(
        self,
        done_indicator: str = "C3560CX#",
        paginate_indicator: str = "--More--",
        paginate_response: str = "\x1d",
    ):
        self.buf = ""
        self.done_indicator = done_indicator
        self.paginate_indicator = paginate_indicator
        self.paginate_response = paginate_response

    def callbacks(self) -> list[ReadCallback]:
        return [
            ReadCallback(
                contains=self.paginate_indicator,
                case_insensitive=False,
                callback=self.paginate_callback,
                name="pagination",
            ),
            ReadCallback(
                contains=self.done_indicator,
                case_insensitive=False,
                callback=self.complete_callback,
                complete=True,
                name="complete",
            ),
        ]

    def paginate_callback(self, cls: GenericDriver, read_output: str) -> None:
        self.buf += read_output

        cls.channel.write(channel_input=self.paginate_response, redacted=False)

    def complete_callback(self, _: GenericDriver, read_output: str) -> None:
        self.buf += read_output


with GenericDriver(**device) as conn:
    paginator = Paginator()
    result = conn.read_callback(paginator.callbacks(), initial_input="show run")
    print(str(paginator.buf))

[Sciti]

Thats exactly what i did, but looks ways better :) Now try to run in somewhere. It'll finish at start, because command will be executed like that C3560CX# show run
Idk, maybe stripping promt in read_callback() or some logic in Paginator.complete_callback() will do the thing.
Btw, is there a way to stop read_callback() from callback function?

[carlmontanari]

Now try to run in somewhere. It'll finish at start

I did run it. It does not finish at the start. It does exactly what you want. Did you run it? Using 0x20 (space) instead of esc, here is the output printed from the last line of the script above (with all my actual config stripped out of the middle).

python .private/more.py
show run
Building configuration...

Current configuration : 10981 bytes
!
<SNIP>
!
ntp server x.x.x.x prefer
!
end

C3560CX#

Changing the str -> repr above shows where the terminal pagination is handled and the rest of the config continues after the callback handles things.

dns-server x.x.x.x \n --More-- \x08\x08\x08\x08\x08\x08\x08\x08\x08        \x08\x08\x08\x08\x08\x08\x08\x08\x08 lease 0 4\n!\nip dhcp pool

[denisbondar]

Hello, friends!

Thanks to @carlmontanari for providing a great example!

Unfortunately, the documentation, which is always very hard to write, is not completely clear for read_callback (and also a bit behind the code).

To make your example work more properly, inside the paginate_callback callback you need to add a new line to the buffer:

def paginate_callback(self, cls: GenericDriver, read_output: str) -> None:
    self.buf += read_output + "\n"

    cls.channel.write(channel_input=self.paginate_response, redacted=False)

Otherwise, the next line is added to the end of the "pagination hint".

My example of using pagination looks like this (I use my extra layer of drivers for ease of instantiation using just one object (schema) instead of multiple constructor parameters and modified the Paginator class a bit.):

with TpLinkDriver(self.device) as conn:
    paginator = Paginator(
        cli_prompt=conn.get_prompt(),
        paginate_hint="Press any key to continue",
    )
    conn.read_callback(callbacks=paginator.callbacks(), initial_input="show mac address-table address all")

result = FsmParser("tplink/tlsl/fdb").parse(paginator.buf)

And everything works as expected. Although, I haven't tested it on such outputs where the pagination hint is displayed more than once, but I think it should work there too.

For me, this issue has been resolved. However, I'm thinking about using the pagination instance not at the end code level, but at the driver level, so that everything is transparent at the end code level.

[carlmontanari]

pagination hint is displayed more than once, but I think it should work there too

yeah as long as the "once" thing isn't true should work -- my test was w/ a running config on iosxe and it for sure had pagination at least a handful of times.

the documentation, which is always very hard to write, is not completely clear for read_callback

yeah... it was kinda yolo'd in there more recently and didnt get the love it maybe should have. if you have cycles any doc improvements or examples (maybe a full example of the TpLinkDriver? or maybe that can just go in scrapli community and we can link to it?) would be great!

[denisbondar]

if you have cycles any doc improvements or examples (maybe a full example of the TpLinkDriver? or maybe that can just go in scrapli community and we can link to it?) would be great!

I didn't like the declarative approach in scrapli_community, due to the inability to autocomplete and potential bugs. So I decided to extend NetworkDriver (and also GenericDriver) similarly to core drivers. Another reason is that in my project all device parameters are collected in the Device pydantic schema. There is the address, credentials for CLI and SNMP and everything that is needed to establish a connection via any of the protocols. Therefore, my driver constructor takes one mandatory argument - device: Device.

My driver base class NetworkPollerDriver looks like this (similarly for the async driver and for the sync and async pair for GenericDriver.):

from scrapli.driver.generic import ReadCallback
from scrapli.driver.generic.sync_driver import GenericDriver
from scrapli.driver.network.sync_driver import NetworkDriver

from src.schemas import Device

class NetworkPollerDriver(NetworkDriver):
    def __init__(
        self,
        device: Device,
        privilege_levels: Dict[str, PrivilegeLevel],
        . . . .
        paginate_hint: str = "--More--",
        paginate_response: str = "\x20"
    ) -> None:
        if not host:
            host = device.host
        if port is None:
            port = device.telnet.port
        if not auth_username:
            auth_username = device.telnet.username
        if not auth_password:
            auth_password = device.telnet.password
        if not auth_secondary:
            auth_secondary = device.telnet.enable_password
        if not auth_telnet_login_pattern:
            auth_telnet_login_pattern = r"^(.*user( name)?:)|(.*username:)|(.*login:)\s?$"
        super().__init__(
            host=host,
            . . . .
        )
        self.paginate_hint = paginate_hint
        self.paginate_response = paginate_response

    def send_command_paginate(
        self,
        command: str,
        read_delay: float = 0.1,
        read_timeout: float = -1.0
    ) -> str:
        paginator = Paginator(
            cli_prompt=self.get_prompt(),
            paginate_hint=self.paginate_hint,
            paginate_response=self.paginate_response
        )

        self.read_callback(
            callbacks=paginator.callbacks(),
            initial_input=command,
            read_delay=read_delay,
            read_timeout=read_timeout
        )

        return paginator.buf

class Paginator:
    def __init__(
        self,
        cli_prompt: str = r"#",
        paginate_hint: str = "--More--",
        paginate_response: str = "\x20",
    ) -> None:
        self.buf = ""
        self.cli_prompt = cli_prompt
        self.paginate_hint = paginate_hint
        self.paginate_response = paginate_response

    def callbacks(self) -> list[ReadCallback]:
        return [
            ReadCallback(
                contains=self.paginate_hint,
                case_insensitive=True,
                callback=self._paginate_callback,
                complete=False,
                name="pagination",
            ),
            ReadCallback(
                contains_re=self.cli_prompt,
                case_insensitive=True,
                callback=self._complete_callback,
                complete=True,
                name="complete",
            ),
        ]

    def _paginate_callback(self, cls: GenericDriver, read_output: str) -> None:
        self.buf += read_output + "\n"

        cls.channel.write(channel_input=self.paginate_response)

    def _complete_callback(self, _: GenericDriver, read_output: str) -> None:
        self.buf += read_output

The concrete classes are very similar to the classes from scraply.driver.core. Using the example of the same TpLinkDriver:

from .base_driver import (
    COMMS_RETURN_CHAR,
    DEFAULT_PRIVILEGE_LEVEL,
    FAILED_WHEN_CONTAINS,
    PAGINATE_HINT,
    PRIVS,
)

def default_on_open(conn: NetworkPollerDriver) -> None:
    conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
    conn.send_config(config="no clipaging")
    conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)


def default_on_close(conn: NetworkPollerDriver) -> None:
    conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
    conn.channel.write(channel_input="exit")
    conn.channel.send_return()


class TpLinkDriver(NetworkPollerDriver):
    def __init__(self, device: Device, **kwargs: Any) -> None:
        if kwargs.get("privilege_levels") is None:
            kwargs["privilege_levels"] = deepcopy(PRIVS)
        if kwargs.get("default_desired_privilege_level") is None:
            kwargs["default_desired_privilege_level"] = DEFAULT_PRIVILEGE_LEVEL

        if kwargs.get("on_open") is None:
            kwargs["on_open"] = default_on_open
        if kwargs.get("on_close") is None:
            kwargs["on_close"] = default_on_close

        if kwargs.get("failed_when_contains") is None:
            kwargs["failed_when_contains"] = FAILED_WHEN_CONTAINS.copy()

        if kwargs.get("comms_return_char") is None:
            kwargs["comms_return_char"] = COMMS_RETURN_CHAR

        if kwargs.get("paginate_hint") is None:
            kwargs["paginate_hint"] = PAGINATE_HINT

        super().__init__(device, **kwargs)

base_driver module contains mixins:

PRIVS = {
    "exec": (PrivilegeLevel...),
    "privilege_exec": (PrivilegeLevel...),
    "configuration": (PrivilegeLevel...),
}
DEFAULT_PRIVILEGE_LEVEL = "privilege_exec"
FAILED_WHEN_CONTAINS = [
   ....
]
COMMS_RETURN_CHAR = "\r\n"
PAGINATE_HINT = "press any key to continue"

As you can see, I've put pagination in the driver base class, and the pagination hint is configured via constants like the others.

My client code is now clean and concise:

with TpLinkDriver(self.device) as conn:
    response = conn.send_command_paginate("show mac address-table address all")

result = FsmParser("tplink/tlsl/fdb").parse(response)

[Sciti]

Wow. Looking good, can i snitch some of your code into my project?

[denisbondar]

Of course!